from collections import defaultdict
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Set, Union

import orjson
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import ValidationError
from django.db import transaction
from django.http import HttpRequest, HttpResponse
from django.utils.translation import gettext as _
from django.utils.translation import override as override_language

from zerver.context_processors import get_valid_realm_from_request
from zerver.decorator import (
    authenticated_json_view,
    require_non_guest_user,
    require_post,
    require_realm_admin,
)
from zerver.lib.actions import (
    bulk_add_subscriptions,
    bulk_remove_subscriptions,
    do_add_default_stream,
    do_add_streams_to_default_stream_group,
    do_change_default_stream_group_description,
    do_change_default_stream_group_name,
    do_change_stream_description,
    do_change_stream_message_retention_days,
    do_change_stream_permission,
    do_change_stream_post_policy,
    do_change_subscription_property,
    do_create_default_stream_group,
    do_deactivate_stream,
    do_delete_messages,
    do_get_streams,
    do_remove_default_stream,
    do_remove_default_stream_group,
    do_remove_streams_from_default_stream_group,
    do_rename_stream,
    do_send_messages,
    gather_subscriptions,
    get_default_streams_for_realm,
    get_subscriber_ids,
    internal_prep_private_message,
    internal_prep_stream_message,
)
from zerver.lib.exceptions import (
    ErrorCode,
    JsonableError,
    OrganizationOwnerRequired,
    ResourceNotFoundError,
)
from zerver.lib.mention import MentionBackend, silent_mention_syntax_for_user
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.retention import parse_message_retention_days
from zerver.lib.streams import (
    StreamDict,
    access_default_stream_group_by_id,
    access_stream_by_id,
    access_stream_by_name,
    access_stream_for_delete_or_update,
    access_web_public_stream,
    check_stream_name_available,
    filter_stream_authorization,
    list_to_streams,
)
from zerver.lib.string_validation import check_stream_name
from zerver.lib.topic import (
    get_topic_history_for_public_stream,
    get_topic_history_for_stream,
    messages_for_topic,
)
from zerver.lib.types import Validator
from zerver.lib.utils import assert_is_not_none
from zerver.lib.validator import (
    check_bool,
    check_capped_string,
    check_color,
    check_dict,
    check_dict_only,
    check_int,
    check_int_in,
    check_list,
    check_string,
    check_string_or_int,
    check_union,
    to_non_negative_int,
)
from zerver.models import (
    Realm,
    Stream,
    UserMessage,
    UserProfile,
    get_active_user,
    get_active_user_profile_by_id_in_realm,
    get_system_bot,
)


class PrincipalError(JsonableError):
    code = ErrorCode.UNAUTHORIZED_PRINCIPAL
    data_fields = ["principal"]
    http_status_code = 403

    def __init__(self, principal: Union[int, str]) -> None:
        self.principal: Union[int, str] = principal

    @staticmethod
    def msg_format() -> str:
        return _("User not authorized to execute queries on behalf of '{principal}'")


def principal_to_user_profile(agent: UserProfile, principal: Union[str, int]) -> UserProfile:
    try:
        if isinstance(principal, str):
            return get_active_user(principal, agent.realm)
        else:
            return get_active_user_profile_by_id_in_realm(principal, agent.realm)
    except UserProfile.DoesNotExist:
        # We have to make sure we don't leak information about which users
        # are registered for Zulip in a different realm.  We could do
        # something a little more clever and check the domain part of the
        # principal to maybe give a better error message
        raise PrincipalError(principal)


def check_if_removing_someone_else(
    user_profile: UserProfile, principals: Optional[Union[List[str], List[int]]]
) -> bool:
    if principals is None or len(principals) == 0:
        return False

    if len(principals) > 1:
        return True

    if isinstance(principals[0], int):
        return principals[0] != user_profile.id
    else:
        return principals[0] != user_profile.email


def deactivate_stream_backend(
    request: HttpRequest, user_profile: UserProfile, stream_id: int
) -> HttpResponse:
    (stream, sub) = access_stream_for_delete_or_update(user_profile, stream_id)
    do_deactivate_stream(stream, acting_user=user_profile)
    return json_success(request)


@require_realm_admin
@has_request_variables
def add_default_stream(
    request: HttpRequest, user_profile: UserProfile, stream_id: int = REQ(json_validator=check_int)
) -> HttpResponse:
    (stream, sub) = access_stream_by_id(user_profile, stream_id)
    if stream.invite_only:
        raise JsonableError(_("Private streams cannot be made default."))
    do_add_default_stream(stream)
    return json_success(request)


@require_realm_admin
@has_request_variables
def create_default_stream_group(
    request: HttpRequest,
    user_profile: UserProfile,
    group_name: str = REQ(),
    description: str = REQ(),
    stream_names: List[str] = REQ(json_validator=check_list(check_string)),
) -> HttpResponse:
    streams = []
    for stream_name in stream_names:
        (stream, sub) = access_stream_by_name(user_profile, stream_name)
        streams.append(stream)
    do_create_default_stream_group(user_profile.realm, group_name, description, streams)
    return json_success(request)


@require_realm_admin
@has_request_variables
def update_default_stream_group_info(
    request: HttpRequest,
    user_profile: UserProfile,
    group_id: int,
    new_group_name: Optional[str] = REQ(default=None),
    new_description: Optional[str] = REQ(default=None),
) -> HttpResponse:
    if not new_group_name and not new_description:
        raise JsonableError(_('You must pass "new_description" or "new_group_name".'))

    group = access_default_stream_group_by_id(user_profile.realm, group_id)
    if new_group_name is not None:
        do_change_default_stream_group_name(user_profile.realm, group, new_group_name)
    if new_description is not None:
        do_change_default_stream_group_description(user_profile.realm, group, new_description)
    return json_success(request)


@require_realm_admin
@has_request_variables
def update_default_stream_group_streams(
    request: HttpRequest,
    user_profile: UserProfile,
    group_id: int,
    op: str = REQ(),
    stream_names
: List[str] = REQ(json_validator=check_list(check_string)),
) -> HttpResponse:
    group = access_default_stream_group_by_id(user_profile.realm, group_id)
    streams = []
    for stream_name in stream_names:
        (strea
m, sub) = access_stream_by_name(user_profile, stream_name)
        streams.append(stream)

    if op == "add":
        do_add_streams_to_default_stream_group(user_profile.realm, group, streams)
    elif op == "remove":
        do_remove_streams_from_default_stream_group(user_profile.realm, group, streams)
    else:
        raise JsonableError(_('Invalid value for "op". Specify one of "add" or "remove".'))
    return json_success(request)


@require_realm_admin
@has_request_variables
def remove_default_stream_group(
    request: HttpRequest, user_profile: UserProfile, group_id: int
) -> HttpResponse:
    group = access_default_stream_group_by_id(user_profile.realm, group_id)
    do_remove_default_stream_group(user_profile.realm, group)
    return json_success(request)


@require_realm_admin
@has_request_variables
def remove_default_stream(
    request: HttpRequest, user_profile: UserProfile, stream_id: int = REQ(json_validator=check_int)
) -> HttpResponse:
    (stream, sub) = access_stream_by_id(
        user_profile,
        stream_id,
        allow_realm_admin=True,
    )
    do_remove_default_stream(stream)
    return json_success(request)


@has_request_variables
def update_stream_backend(
    request: HttpRequest,
    user_profile: UserProfile,
    stream_id: int,
    description: Optional[str] =