from collections import defaultdict
from collections.abc import Collection, Iterable
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, TypedDict

from django.db import transaction
from django.db.models import Exists, OuterRef, Q, QuerySet
from django.utils.timezone import now as timezone_now
from django.utils.translation import gettext as _

from zerver.lib.default_streams import get_default_stream_ids_for_realm
from zerver.lib.exceptions import (
    CannotAdministerChannelError,
    CannotSetTopicsPolicyError,
    ChannelExistsError,
    IncompatibleParametersError,
    JsonableError,
    OrganizationOwnerRequiredError,
)
from zerver.lib.stream_subscription import (
    get_guest_user_ids_for_streams,
    get_subscribed_stream_ids_for_user,
    get_user_ids_for_streams,
)
from zerver.lib.stream_traffic import get_average_weekly_stream_traffic, get_streams_traffic
from zerver.lib.string_validation import check_stream_name
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.lib.topic import get_topic_display_name, messages_for_topic
from zerver.lib.types import APIStreamDict, UserGroupMembersData
from zerver.lib.user_groups import (
    UserGroupMembershipDetails,
    access_user_group_for_setting,
    get_group_setting_value_for_register_api,
    get_members_and_subgroups_of_groups,
    get_recursive_membership_groups,
    get_role_based_system_groups_dict,
    get_root_id_annotated_recursive_subgroups_for_groups,
    parse_group_setting_value,
    user_has_permission_for_group_setting,
)
from zerver.models import (
    ChannelFolder,
    DefaultStreamGroup,
    Message,
    NamedUserGroup,
    Realm,
    RealmAuditLog,
    Recipient,
    Stream,
    Subscription,
    UserGroup,
    UserGroupMembership,
    UserProfile,
)
from zerver.models.groups import SystemGroups
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.streams import (
    StreamTopicsPolicyEnum,
    bulk_get_streams,
    get_realm_stream,
    get_stream,
    get_stream_by_id_for_sending_message,
    get_stream_by_id_in_realm,
)
from zerver.models.users import active_non_guest_user_ids, active_user_ids, is_cross_realm_bot_email
from zerver.tornado.django_api import send_event_on_commit


class StreamDict(TypedDict, total=False):
    """
    This type ultimately gets used in two places:

        - we use it to create a stream
        - we use it to specify a stream


    It's possible we want a smaller type to use
    for removing streams, but it would complicate
    how we write the types for list_to_stream.

    Note that these fields are just a subset of
    the fields in the Stream model.
    """

    name: str
    description: str
    invite_only: bool
    is_web_public: bool
    stream_post_policy: int
    history_public_to_subscribers: bool | None
    message_retention_days: int | None
    topics_policy: int | None
    can_add_subscribers_group: UserGroup | None
    can_administer_channel_group: UserGroup | None
    can_create_topic_group: UserGroup | None
    can_delete_any_message_group: UserGroup | None
    can_delete_own_message_group: UserGroup | None
    can_move_messages_out_of_channel_group: UserGroup | None
    can_move_messages_within_channel_group: UserGroup | None
    can_send_message_group: UserGroup | None
    can_remove_subscribers_group: UserGroup | None
    can_resolve_topics_group: UserGroup | None
    can_subscribe_group: UserGroup | None
    folder: ChannelFolder | None


def get_stream_permission_policy_key(
    *,
    invite_only: bool | None = None,
    history_public_to_subscribers: bool | None = None,
    is_web_public: bool | None = None,
) -> str:
    policy_key = None
    for permission_key, permission_dict in Stream.PERMISSION_POLICIES.items():
        if (
            permission_dict["invite_only"] == invite_only
            and permission_dict["history_public_to_subscribers"] == history_public_to_subscribers
            and permission_dict["is_web_public"] == is_web_public
        ):
            policy_key = permission_key
            break

    assert policy_key is not None
    return policy_key


def get_stream_topics_policy(realm: Realm, stream: Stream) -> int:
    if stream.topics_policy == StreamTopicsPolicyEnum.inherit.value:
        return realm.topics_policy
    return stream.topics_policy


def validate_topics_policy(
    topics_policy: str | None,
    user_profile: UserProfile,
    # Pass None when creating a channel and the channel being edited when editing a channel's settings
    stream: Stream | None = None,
) -> StreamTopicsPolicyEnum | None:
    if topics_policy is not None and isinstance(topics_policy, StreamTopicsPolicyEnum):
        if (
            topics_policy != StreamTopicsPolicyEnum.inherit
            and not user_profile.can_set_topics_policy()
        ):
            raise JsonableError(_("Insufficient permission"))

        # Cannot set `topics_policy` to `empty_topic_only` when there are messages
        # in non-empty topics in the current channel.
        if (
            stream is not None
            and topics_policy == StreamTopicsPolicyEnum.empty_topic_only
            and channel_has_named_topics(stream)
        ):
            raise CannotSetTopicsPolicyError(
                get_topic_display_name("", user_profile.default_language)
            )
        return topics_policy
    return None


def validate_can_create_topic_group_setting_for_protected_history_streams(
    history_public_to_subscribers: bool | None,
    invite_only: bool,
    can_create_topic_group: int | UserGroupMembersData,
    system_groups_name_dict: dict[str, NamedUserGroup],
) -> None:
    if history_public_to_subscribers is None:
        history_public_to_subscribers = get_default_value_for_history_public_to_subscribers(
            invite_only, history_public_to_subscribers
        )

    if history_public_to_subscribers:
        return

    # If a group setting has value "Everyone including guests" along with additional
    # users or groups, we do not treat it as equivalent to just "Everyone including guests".
    # For a channel with protected history, everyone must be allowed to create new topics.
    # As a result, enabling protected history for a channel requires the `can_create_topic_group`
    # setting to have "Everyone including guests" group configuration only.
    if (
        not isinstance(can_create_topic_group, int)
        or can_create_topic_group != system_groups_name_dict[SystemGroups.EVERYONE].id
    ):
        raise IncompatibleParametersError(
            ["history_public_to_subscribers", "can_create_topic_group"]
        )


def get_default_value_for_history_public_to_subscribers(
    invite_only: bool,
    history_public_to_subscribers: bool | None,
) -> bool:
    if invite_only:
        if history_public_to_subscribers is None:
            # A private stream's history is non-public by default
            history_public_to_subscribers = False
    else:
        # If we later decide to support public streams without
        # history, we can remove this code path.
        history_public_to_subscribers = True

    return history_public_to_subscribers


def render_stream_description(
    text: str, realm: Realm, *, acting_user: UserProfile | None = None
) -> str:
    from zerver.lib.markdown import markdown_convert

    return markdown_convert(
        text, message_realm=realm, no_previews=True, acting_user=acting_user
    ).rendered_content


def send_stream_creation_event(
    realm: Realm,
    stream: Stream,
    user_ids: list[int],
    recent_traffic: dict[int, int] | None = None,
    anonymous_group_membership: dict[int, UserGroupMembersData] | None = None,
    for_unarchiving: bool = False,
) -> None:
    event = dict(
        type="stream",
        op="create",
        streams=[stream_to_dict(stream, recent_traffic, anonymous_group_membership)],
        for_unarchiving=for_unarchiving,
    )
    send_event_on_commit(realm, event, user_ids)


def check_channel_creation_permissions(
    user_profile: UserProfile,
    *,
    is_default_stream: bool,
    invite_only: bool,
    is_web_public: bool,
    message_retention_days: str | int | None,
) -> None:
    if invite_only and not user_profile.can_create_private_streams():
        raise JsonableError(_("Insufficient permission"))
    if not invite_only and not user_profile.can_create_public_streams():
        raise JsonableError(_("Insufficient permission"))
    if is_default_stream and not user_profile.is_realm_admin:
        raise JsonableError(_("Insufficient permission"))
    if invite_only and is_default_stream:
        raise JsonableError(_("A default channel cannot be private."))
    if is_web_public:
        if not user_profile.realm.web_public_streams_enabled():
            raise JsonableError(_("Web-public channels are not enabled."))
        if not user_profile.can_create_web_public_streams():
            # We set can_create_web_public_channel_group to allow only organization
            # owners to create web-public streams, because of their sensitive nature.
            raise JsonableError(_("Insufficient permission"))
    if message_retention_days is not None:
        if not user_profile.is_realm_owner:
            raise OrganizationOwnerRequiredError
        user_profile.realm.ensure_not_on_limited_plan()


def get_stream_permission_default_group(
    setting_name: str,
    system_groups_name_dict: dict[str, NamedUserGroup],
    creator: UserProfile | None = None,
) -> UserGroup:
    setting_default_name = Stream.stream_permission_group_settings[setting_name].default_group_name
    if setting_default_name == "channel_creator":
        if creator:
            default_group = UserGroup(
                realm=creator.realm,
            )
            default_group.save()
            UserGroupMembership.objects.create(user_profile=creator, user_group=default_group)
            return default_group
        else:
            return system_groups_name_dict[SystemGroups.NOBODY]
    return system_groups_name_dict[setting_default_name]


def get_default_values_for_stream_permission_group_settings(
    realm: Realm, creator: UserProfile | None = None
) -> dict[str, UserGroup]:
    group_setting_values = {}
    system_groups_name_dict = get_role_based_system_groups_dict(realm)
    for setting_name in Stream.stream_permission_group_settings:
        group_setting_values[setting_name] = get_stream_permission_default_group(
            setting_name, system_groups_name_dict, creator
        )

    return group_setting_values


def get_users_dict_with_metadata_access_to_streams_via_permission_groups(
    streams: list[Stream],
    realm_id: int,
) -> dict[int, set[int]]:
    can_administer_group_ids = {stream.can_administer_channel_group_id for stream in streams}
    can_add_subscriber_group_ids = {stream.can_add_subscribers_group_id for stream in streams}
    can_subscribe_group_ids = {stream.can_subscribe_group_id for stream in streams}

    all_permission_group_ids = list(
        can_administer_group_ids | can_add_subscriber_group_ids | can_subscribe_group_ids
    )

    recursive_subgroups = get_root_id_annotated_recursive_subgroups_for_groups(
        all_permission_group_ids, realm_id
    )
    subgroup_root_id_dict = {}
    all_subgroup_ids = set()
    for group in recursive_subgroups:
        subgroup_root_id_dict[group.id] = group.root_id  # type: ignore[attr-defined]  # root_id is an annotated field.
        all_subgroup_ids.add(group.id)

    group_members = (
        UserGroupMembership.objects.filter(
            user_group_id__in=list(all_subgroup_ids), user_profile__is_active=True
        )
        .exclude(
            # allow_everyone_group=False is false for both
            # can_add_subscribers_group and
            # can_administer_channel_group, so guest users cannot
            # exercise these permission to get metadata access.
            user_profile__role=UserProfile.ROLE_GUEST
        )
        .values_list("user_group_id", "user_profile_id")
    )
    group_members_dict = defaultdict(set)
    for user_group_id, user_profile_id in group_members:
        root_id = subgroup_root_id_dict[user_group_id]
        group_members_dict[root_id].add(user_profile_id)

    users_with_metadata_access_dict = defaultdict(set)
    for stream in streams:
        users_with_metadata_access_dict[stream.id] = (
            group_members_dict[stream.can_administer_channel_group_id]
            | group_members_dict[stream.can_add_subscribers_group_id]
            | group_members_dict[stream.can_subscribe_group_id]
        )

    return users_with_metadata_access_dict


def get_user_ids_with_metadata_access_via_permission_groups(stream: Stream) -> set[int]:
    users_with_metadata_access_dict = (
        get_users_dict_with_metadata_access_to_streams_via_permission_groups(
            [stream], stream.realm_id
        )
    )
    return users_with_metadata_access_dict[stream.id]


def channel_events_topic_name(stream: Stream) -> str:
    if stream.topics_policy == StreamTopicsPolicyEnum.empty_topic_only.value:
        return ""
    return str(Realm.STREAM_EVENTS_NOTIFICATION_TOPIC_NAME)


def channel_has_named_topics(stream: Stream) -> bool:
    return (
        Message.objects.filter(realm_id=stream.realm_id, recipient=stream.recipient)
        .exclude(subject="")
        .exists()
    )


@transaction.atomic(savepoint=False)
def create_stream_if_needed(
    realm: Realm,
    stream_name: str,
    *,
    invite_only: bool = False,
    is_web_public: bool = False,
    history_public_to_subscribers: bool | None = None,
    stream_description: str = "",
    message_retention_days: int | None = None,
    topics_policy: int | None = None,
    can_add_subscribers_group: UserGroup | None = None,
    can_administer_channel_group: UserGroup | None = None,
    can_create_topic_group: UserGroup | None = None,
    can_delete_any_message_group: UserGroup | None = None,
    can_delete_own_message_group: UserGroup | None = None,
    can_move_messages_out_of_channel_group: UserGroup | None = None,
    can_move_messages_within_channel_group: UserGroup | None = None,
    can_send_message_group: UserGroup | None = None,
    can_remove_subscribers_group: UserGroup | None = None,
    can_resolve_topics_group: UserGroup | None = None,
    can_subscribe_group: UserGroup | None = None,
    folder: ChannelFolder | None = None,
    acting_user: UserProfile | None = None,
    anonymous_group_membership: dict[int, UserGroupMembersData] | None = None,
) -> tuple[Stream, bool]:
    history_public_to_subscribers = get_default_value_for_history_public_to_subscribers(
        invite_only, history_public_to_subscribers
    )

    group_setting_values = {}
    request_settings_dict = locals()
    # We don't want to calculate this value if no default values are
    # needed.
    system_groups_name_dict = None
    for setting_name in Stream.stream_permission_group_settings:
        if setting_name not in request_settings_dict:  # nocoverage
            continue

        if request_settings_dict[setting_name] is None:
            if system_groups_name_dict is None:
                system_groups_name_dict = get_role_based_system_groups_dict(realm)
            group_setting_values[setting_name] = get_stream_permission_default_group(
                setting_name, system_groups_name_dict, creator=acting_user
            )
        else:
            group_setting_values[setting_name] = request_settings_dict[setting_name]

    stream_name = stream_name.strip()

    if topics_policy is None:
        topics_policy = StreamTopicsPolicyEnum.inherit.value

    (stream, created) = Stream.objects.get_or_create(
        realm=realm,
        name__iexact=stream_name,
        defaults=dict(
            name=stream_name,
            creator=acting_user,
            description=stream_description,
            invite_only=invite_only,
            is_web_public=is_web_public,
            history_public_to_subscribers=history_public_to_subscribers,
            message_retention_days=message_retention_days,
            folder=folder,
            topics_policy=topics_policy,
            **group_setting_values,
        ),
    )

    if created:
        recipient = Recipient.objects.create(type_id=stream.id, type=Recipient.STREAM)

        stream.recipient = recipient
        stream.rendered_description = render_stream_description(
            stream_description, realm, acting_user=acting_user
        )
        stream.save(update_fields=["recipient", "rendered_description"])

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

        if anonymous_group_membership is None:
            anonymous_group_membership = get_anonymous_group_membership_dict_for_streams([stream])

        if stream.is_public():
            if stream.is_web_public:
                notify_user_ids = active_user_ids(stream.realm_id)
            else:
                # TODO: This should include guests with metadata
                # access to the channel, once that is possible via
                # can_join_group.
                notify_user_ids = active_non_guest_user_ids(stream.realm_id)
            send_stream_creation_event(
                realm,
                stream,
                notify_user_ids,
                anonymous_group_membership=anonymous_group_membership,
            )
        else:
            realm_admin_ids = {user.id for user in stream.realm.get_admin_users_and_bots()}
            send_stream_creation_event(
                realm,
                stream,
                list(
                    realm_admin_ids
                    | get_user_ids_with_metadata_access_via_permission_groups(stream)
                ),
                anonymous_group_membership=anonymous_group_membership,
            )

    return stream, created


def create_streams_if_needed(
    realm: Realm,
    stream_dicts: list[StreamDict],
    acting_user: UserProfile | None = None,
    anonymous_group_membership: dict[int, UserGroupMembersData] | None = None,
) -> tuple[list[Stream], list[Stream]]:
    """Note that stream_dict["name"] is assumed to already be stripped of
    whitespace"""
    added_streams: list[Stream] = []
    existing_streams: list[Stream] = []
    for stream_dict in stream_dicts:
        invite_only = stream_dict.get("invite_only", False)
        stream, created = create_stream_if_needed(
            realm,
            stream_dict["name"],
            invite_only=invite_only,
            is_web_public=stream_dict.get("is_web_public", False),
            history_public_to_subscribers=stream_dict.get("history_public_to_subscribers"),
            stream_description=stream_dict.get("description", ""),
            message_retention_days=stream_dict.get("message_retention_days", None),
            topics_policy=stream_dict.get("topics_policy", None),
            can_add_subscribers_group=stream_dict.get("can_add_subscribers_group", None),
            can_administer_channel_group=stream_dict.get("can_administer_channel_group", None),
            can_create_topic_group=stream_dict.get("can_create_topic_group", None),
            can_delete_any_message_group=stream_dict.get("can_delete_any_message_group", None),
            can_delete_own_message_group=stream_dict.get("can_delete_own_message_group", None),
            can_move_messages_out_of_channel_group=stream_dict.get(
                "can_move_messages_out_of_channel_group", None
            ),
            can_move_messages_within_channel_group=stream_dict.get(
                "can_move_messages_within_channel_group", None
            ),
            can_send_message_group=stream_dict.get("can_send_message_group", None),
            can_remove_subscribers_group=stream_dict.get("can_remove_subscribers_group", None),
            can_resolve_topics_group=stream_dict.get("can_resolve_topics_group", None),
            can_subscribe_group=stream_dict.get("can_subscribe_group", None),
            folder=stream_dict.get("folder", None),
            acting_user=acting_user,
            anonymous_group_membership=anonymous_group_membership,
        )

        if created:
            added_streams.append(stream)
        else:
            existing_streams.append(stream)

    return added_streams, existing_streams


def subscribed_to_stream(user_profile: UserProfile, stream_id: int) -> bool:
    return Subscription.objects.filter(
        user_profile=user_profile,
        active=True,
        recipient__type=Recipient.STREAM,
        recipient__type_id=stream_id,
    ).exists()


def is_user_in_can_administer_channel_group(
    stream: Stream, user_recursive_group_ids: set[int]
) -> bool:
    # Important: The caller must have verified the acting user is not
    # a guest, to enforce that can_administer_channel_group has
    # allow_everyone_group=False.
    group_allowed_to_administer_channel_id = stream.can_administer_channel_group_id
    assert group_allowed_to_administer_channel_id is not None
    return group_allowed_to_administer_channel_id in user_recursive_group_ids


def is_user_in_can_add_subscribers_group(
    stream: Stream, user_recursive_group_ids: set[int]
) -> bool:
    # Important: The caller must have verified the acting user is not
    # a guest, to enforce that can_add_subscribers_group has
    # allow_everyone_group=False.
    group_allowed_to_add_subscribers_id = stream.can_add_subscribers_group_id
    assert group_allowed_to_add_subscribers_id is not None
    return group_allowed_to_add_subscribers_id in user_recursive_group_ids


def is_user_in_can_subscribe_group(stream: Stream, user_recursive_group_ids: set[int]) -> bool:
    # Important: The caller must have verified the acting user
    # is not a guest, to enforce that can_subscribe_group has
    # allow_everyone_group=False.
    group_allowed_to_subscribe_id = stream.can_subscribe_group_id
    return group_allowed_to_subscribe_id in user_recursive_group_ids


def is_user_in_groups_granting_content_access(
    stream: Stream, user_recursive_group_ids: set[int]
) -> bool:
    # Important: The caller must have verified the acting user is not
    # a guest, to enforce that can_add_subscribers_group has
    # allow_everyone_group=False.
    return is_user_in_can_subscribe_group(
        stream, user_recursive_group_ids
    ) or is_user_in_can_add_subscribers_group(stream, user_recursive_group_ids)


def is_user_in_can_remove_subscribers_group(
    stream: Stream, user_recursive_group_ids: set[int]
) -> bool:
    # Important: The caller must have verified the acting user is not
    # a guest, to enforce that can_remove_subscribers_group has
    # allow_everyone_group=False.
    group_allowed_to_remove_subscribers_id = stream.can_remove_subscribers_group_id
    assert group_allowed_to_remove_subscribers_id is not None
    return group_allowed_to_remove_subscribers_id in user_recursive_group_ids


def check_stream_access_based_on_can_send_message_group(
    sender: UserProfile, stream: Stream
) -> None:
    if is_cross_realm_bot_email(sender.delivery_email):
        return

    can_send_message_group = stream.can_send_message_group
    if hasattr(can_send_message_group, "named_user_group"):
        if can_send_message_group.named_user_group.name == SystemGroups.EVERYONE:
            return

        if can_send_message_group.named_user_group.name == SystemGroups.NOBODY:
            raise JsonableError(_("You do not have permission to post in this channel."))

    if not user_has_permission_for_group_setting(
        stream.can_send_message_group_id,
        sender,
        Stream.stream_permission_group_settings["can_send_message_group"],
        direct_member_only=False,
    ):
        raise JsonableError(_("You do not have permission to post in this channel."))


def access_stream_for_send_message(
    sender: UserProfile,
    stream: Stream,
    forwarder_user_profile: UserProfile | None,
    archived_channel_notice: bool = False,
) -> None:
    # Our caller is responsible for making sure that `stream` actually
    # matches the realm of the sender.
    try:
        check_stream_access_based_on_can_send_message_group(sender, stream)
    except JsonableError as e:
        if sender.is_bot and sender.bot_owner is not None:
            check_stream_access_based_on_can_send_message_group(sender.bot_owner, stream)
        else:
            raise JsonableError(e.msg)

    # forwarder_user_profile cases should be analyzed first, as incorrect
    # message forging is cause for denying access regardless of any other factors.
    if forwarder_user_profile is not None and forwarder_user_profile != sender:
        if (
            forwarder_user_profile.can_forge_sender
            and forwarder_user_profile.realm_id == sender.realm_id
            and sender.realm_id == stream.realm_id
        ):
            return
        else:
            raise JsonableError(_("User not authorized for this query"))

    # You cannot send mesasges to archived channels
    if stream.deactivated:
        if archived_channel_notice:
            return
        raise JsonableError(
            _("Not authorized to send to channel '{channel_name}'").format(channel_name=stream.name)
        )

    if is_cross_realm_bot_email(sender.delivery_email):
        return

    if stream.realm_id != sender.realm_id:
        # Sending to other realm's streams is always disallowed,
        # with the exception of cross-realm bots.
        raise JsonableError(_("User not authorized for this query"))

    if stream.is_web_public:
        # Even guest users can write to web-public streams.
        return

    if not (stream.invite_only or sender.is_guest):
        # This is a public stream and sender is not a guest user
        return

    if subscribed_to_stream(sender, stream.id):
        # It is private, but your are subscribed
        return

    if sender.can_forge_sender:
        # can_forge_sender allows sending to any stream in the realm.
        return

    if sender.is_bot and (
        sender.bot_owner is not None and subscribed_to_stream(sender.bot_owner, stream.id)
    ):
        # Bots can send to any stream their owner can.
        return

    user_recursive_group_ids = set(
        get_recursive_membership_groups(sender).values_list("id", flat=True)
    )

    if (
        stream.history_public_to_subscribers
        and is_user_in_groups_granting_content_access(stream, user_recursive_group_ids)
        and not sender.is_guest
    ):
        return

    # All other cases are an error.
    raise JsonableError(
        _("Not authorized to send to channel '{channel_name}'").format(channel_name=stream.name)
    )


def check_user_can_create_new_topics(user_profile: UserProfile, stream: Stream) -> None:
    can_create_topic_group = stream.can_create_topic_group
    if (
        hasattr(can_create_topic_group, "named_user_group")
        and can_create_topic_group.named_user_group.name == SystemGroups.NOBODY
    ):
        raise JsonableError(_("You do not have permission to create new topics in this channel."))

    if not user_has_permission_for_group_setting(
        stream.can_create_topic_group_id,
        user_profile,
        Stream.stream_permission_group_settings["can_create_topic_group"],
        direct_member_only=False,
    ):
        raise JsonableError(_("You do not have permission to create new topics in this channel."))


def check_for_can_create_topic_group_violation(
    user_profile: UserProfile, stream: Stream, topic_name: str
) -> None:
    if is_cross_realm_bot_email(user_profile.delivery_email):
        return

    can_create_topic_group = stream.can_create_topic_group
    if (
        hasattr(can_create_topic_group, "named_user_group")
        and can_create_topic_group.named_user_group.name == SystemGroups.EVERYONE
    ):
        return

    assert stream.recipient_id is not None
    topic_exists = messages_for_topic(
        realm_id=stream.realm_id, stream_recipient_id=stream.recipient_id, topic_name=topic_name
    ).exists()

    if not topic_exists:
        try:
            check_user_can_create_new_topics(user_profile, stream)
        except JsonableError as e:
            if user_profile.is_bot and user_profile.bot_owner is not None:
                check_user_can_create_new_topics(user_profile.bot_owner, stream)
            else:
                raise JsonableError(e.msg)


def check_for_exactly_one_stream_arg(stream_id: int | None, stream: str | None) -> None:
    if stream_id is None and stream is None:
        # Uses the same translated string as RequestVariableMissingError
        # with the stream_id parameter, which is the more common use case.
        error = _("Missing '{var_name}' argument").format(var_name="stream_id")
        raise JsonableError(error)

    if stream_id is not None and stream is not None:
        raise IncompatibleParametersError(["stream_id", "stream"])


def user_has_metadata_access(
    user_profile: UserProfile,
    stream: Stream,
    user_group_membership_details: UserGroupMembershipDetails,
    *,
    is_subscribed: bool,
) -> bool:
    if stream.is_web_public:
        return True

    if is_subscribed:
        return True

    if user_profile.is_guest:
        return False

    if stream.is_public():
        return True

    if user_profile.is_realm_admin:
        return True

    if user_group_membership_details.user_recursive_group_ids is None:
        user_group_membership_details.user_recursive_group_ids = set(
            get_recursive_membership_groups(user_profile).values_list("id", flat=True)
        )

    if has_metadata_access_to_channel_via_groups(
        user_profile,
        user_group_membership_details.user_recursive_group_ids,
        stream.can_administer_channel_group_id,
        stream.can_add_subscribers_group_id,
        stream.can_subscribe_group_id,
    ):
        return True

    return False


def user_has_content_access(
    user_profile: UserProfile,
    stream: Stream,
    user_group_membership_details: UserGroupMembershipDetails,
    *,
    is_subscribed: bool,
) -> bool:
    if stream.is_web_public:
        return True

    if is_subscribed:
        return True

    if user_profile.is_guest:
        return False

    if stream.is_public():
        return True

    if user_group_membership_details.user_recursive_group_ids is None:
        user_group_membership_details.user_recursive_group_ids = set(
            get_recursive_membership_groups(user_profile).values_list("id", flat=True)
        )

    # This check must be after the user_profile.is_guest check, since
    # allow_everyone_group=False for can_add_subscribers_group and
    # can_subscribe_group.
    if is_user_in_groups_granting_content_access(
        stream, user_group_membership_details.user_recursive_group_ids
    ):
        return True

    return False


def check_stream_access_for_delete_or_update_requiring_metadata_access(
    user_profile: UserProfile, stream: Stream, sub: Subscription | None = None
) -> None:
    error = _("Invalid channel ID")
    if stream.realm_id != user_profile.realm_id:
        raise JsonableError(error)

    # Optimization for the organization administrator code path. We
    # don't explicitly grant realm admins this permission, but admins
    # implicitly have the can_administer_channel_group permission for
    # all accessible channels.
    if user_profile.is_realm_admin:
        return

    if can_administer_accessible_channel(stream, user_profile):
        return

    # We only want to reveal that user is not an administrator
    # if the user has access to the channel in the first place.
    # Ideally, we would be checking if user has metadata access
    # to the channel for this block, but since we have ruled out
    # the possibility that the user is a channel admin, checking
    # for content access will save us valuable DB queries.
    user_group_membership_details = UserGroupMembershipDetails(user_recursive_group_ids=None)
    if user_has_content_access(
        user_profile, stream, user_group_membership_details, is_subscribed=sub is not None
    ):
        raise CannotAdministerChannelError

    raise JsonableError(error)


def access_stream_for_delete_or_update_requiring_metadata_access(
    user_profile: UserProfile, stream_id: int
) -> tuple[Stream, Subscription | None]:
    try:
        stream = Stream.objects.get(id=stream_id)
    except Stream.DoesNotExist:
        raise JsonableError(_("Invalid channel ID"))

    try:
