import logging
from collections import Counter
from datetime import datetime, timedelta, timezone
from email.headerregistry import Address
from email.utils import formatdate as email_formatdate
from typing import Annotated, Any, TypedDict, TypeVar
from urllib.parse import urljoin, urlsplit
from uuid import UUID

import orjson
import requests.exceptions
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.validators import URLValidator, validate_email
from django.db import IntegrityError, transaction
from django.db.models import Model, QuerySet
from django.db.models.constants import OnConflict
from django.db.models.functions import Lower
from django.http import HttpRequest, HttpResponse
from django.utils.crypto import constant_time_compare, get_random_string
from django.utils.timezone import now as timezone_now
from django.utils.translation import gettext as _
from django.utils.translation import gettext as err_
from django.views.decorators.csrf import csrf_exempt
from dns import resolver as dns_resolver
from dns.exception import DNSException
from nacl.encoding import Base64Encoder
from nacl.exceptions import CryptoError
from nacl.public import PrivateKey, SealedBox
from pydantic import BaseModel, ConfigDict, Json, StringConstraints, model_validator
from pydantic import ValidationError as PydanticValidationError
from pydantic.functional_validators import AfterValidator
from typing_extensions import override

from analytics.lib.counts import (
    BOUNCER_ONLY_REMOTE_COUNT_STAT_PROPERTIES,
    COUNT_STATS,
    LOGGING_COUNT_STAT_PROPERTIES_NOT_SENT_TO_BOUNCER,
    REMOTE_INSTALLATION_COUNT_STATS,
    do_increment_logging_stat,
)
from corporate.models.customers import get_customer_by_remote_realm
from corporate.models.plans import CustomerPlan, get_current_plan_by_customer
from zerver.decorator import require_post
from zerver.lib.email_validation import validate_is_not_disposable
from zerver.lib.exceptions import (
    ErrorCode,
    InvalidBouncerPublicKeyError,
    InvalidEncryptedPushRegistrationError,
    JsonableError,
    MissingRemoteRealmError,
    RateLimitedError,
    RemoteRealmServerMismatchError,
    RemoteServerDeactivatedError,
    RequestExpiredError,
)
from zerver.lib.outgoing_http import OutgoingSession
from zerver.lib.push_notifications import (
    PUSH_REGISTRATION_LIVENESS_TIMEOUT,
    APNsPushRequest,
    FCMPushRequest,
    HostnameAlreadyInUseBouncerError,
    InvalidRemotePushDeviceTokenError,
    RealmPushStatusDict,
    UserPushIdentityCompat,
    send_android_push_notification,
    send_apple_push_notification,
    send_test_push_notification_directly_to_devices,
    validate_token,
)
from zerver.lib.queue import queue_event_on_commit
from zerver.lib.rate_limiter import rate_limit_endpoint_absolute
from zerver.lib.remote_server import (
    InstallationCountDataForAnalytics,
    RealmAuditLogDataForAnalytics,
    RealmCountDataForAnalytics,
    RealmDataForAnalytics,
)
from zerver.lib.request import RequestNotes
from zerver.lib.response import json_success
from zerver.lib.send_email import FromAddress
from zerver.lib.timestamp import datetime_to_timestamp, timestamp_to_datetime
from zerver.lib.typed_endpoint import (
    ApnsAppId,
    JsonBodyPayload,
    RequiredStringConstraint,
    typed_endpoint,
    typed_endpoint_without_parameters,
)
from zerver.lib.typed_endpoint_validators import check_string_fixed_length
from zerver.lib.types import RemoteRealmDictValue
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import DisposableEmailError, Realm
from zilencer.auth import (
    InvalidZulipServerKeyError,
    generate_registration_transfer_verification_secret,
    validate_registration_transfer_verification_secret,
)
from zilencer.lib.push_notifications import send_e2ee_push_notifications
from zilencer.lib.remote_counts import MissingDataError
from zilencer.models import (
    RemoteInstallationCount,
    RemotePushDevice,
    RemotePushDeviceToken,
    RemoteRealm,
    RemoteRealmAuditLog,
    RemoteRealmCount,
    RemoteZulipServer,
    RemoteZulipServerAuditLog,
)

logger = logging.getLogger(__name__)


def validate_uuid(uuid: str) -> None:
    try:
        uuid_object = UUID(uuid, version=4)
        # The UUID initialization under some circumstances will modify the uuid
        # string to create a valid UUIDv4, instead of raising a ValueError.
        # The submitted uuid needing to be modified means it's invalid, so
        # we need to check for that condition.
        if str(uuid_object) != uuid:
            raise ValidationError(err_("Invalid UUID"))
    except ValueError:
        raise ValidationError(err_("Invalid UUID"))


def validate_bouncer_token_request(token: str, kind: int) -> None:
    if kind not in [RemotePushDeviceToken.APNS, RemotePushDeviceToken.FCM]:
        raise JsonableError(err_("Invalid token type"))
    validate_token(token, kind)


@csrf_exempt
@require_post
@typed_endpoint_without_parameters
def deactivate_remote_server(
    request: HttpRequest,
    remote_server: RemoteZulipServer,
) -> HttpResponse:
    from corporate.lib.stripe import RemoteServerBillingSession, do_deactivate_remote_server

    billing_session = RemoteServerBillingSession(remote_server)
    do_deactivate_remote_server(remote_server, billing_session)
    return json_success(request)


def validate_hostname_or_raise_error(hostname: str) -> None:
    """
    Validate that the argument is a valid hostname to be registered
    with the bouncer.

    Note: The hostname is allowed to contain a port number, as some
    registration do take this form and without the port we wouldn't
    actually know how to make requests to the server.
    """
    try:
        # We perform basic validation in two steps:
        # 1. urlsplit doesn't do any proper validation, but parses the string
        #    and ensures that there are no extra components (e.g., path, query, fragment).
        # 2. Once we know that the string is a clean netloc, we pass that to Django's
        #    URLValidator for validation.
        parsed = urlsplit(f"http://{hostname}")

        if parsed.path or parsed.query or parsed.fragment:
            raise JsonableError(
                _("{hostname} contains invalid components (e.g., path, query, fragment).").format(
                    hostname=hostname
                )
            )
        url_validator = URLValidator()
        url_validator("http://" + hostname)
    except ValidationError:
        raise JsonableError(_("{hostname} is not a valid hostname").format(hostname=hostname))


@csrf_exempt
@require_post
@typed_endpoint
def transfer_remote_server_registration(request: HttpRequest, *, hostname: str) -> HttpResponse:
    validate_hostname_or_raise_error(hostname)

    if not RemoteZulipServer.objects.filter(hostname=hostname, deactivated=False).exists():
        raise JsonableError(_("{hostname} not yet registered").format(hostname=hostname))

    verification_secret = generate_registration_transfer_verification_secret(hostname)
    return json_success(
        request,
        data={
            "verification_secret": verification_secret,
        },
    )


class ServerAdminEmailError(JsonableError):
    http_status_code = 400
    data_fields = ["email_reason"]

    def __init__(self, email_reason: str) -> None:
        self.email_reason = email_reason

    @staticmethod
    @override
    def msg_format() -> str:
        return _("Invalid server administrator email address: {email_reason}")


@csrf_exempt
@require_post
@typed_endpoint
def register_remote_server(
    request: HttpRequest,
    *,
    zulip_org_id: Annotated[
        str,
        RequiredStringConstraint,
        AfterValidator(lambda s: check_string_fixed_length(s, RemoteZulipServer.UUID_LENGTH)),
    ],
    zulip_org_key: Annotated[
        str,
        RequiredStringConstraint,
        AfterValidator(lambda s: check_string_fixed_length(s, RemoteZulipServer.API_KEY_LENGTH)),
    ],
    hostname: Annotated[str, StringConstraints(max_length=RemoteZulipServer.HOSTNAME_MAX_LENGTH)],
    contact_email: str,
    new_org_key: Annotated[
        str | None,
        RequiredStringConstraint,
        AfterValidator(lambda s: check_string_fixed_length(s, RemoteZulipServer.API_KEY_LENGTH)),
    ] = None,
) -> HttpResponse:
    # StringConstraints validated the field lengths, but we still need to
    # validate the format of these fields.
    validate_hostname_or_raise_error(hostname)

    try:
        validate_email(contact_email)
    except ValidationError as e:
        raise ServerAdminEmailError(str(e.message))

    # We don't want to allow disposable domains for contact_email either
    try:
        validate_is_not_disposable(contact_email)
    except DisposableEmailError:
        raise ServerAdminEmailError(_("Please use your real email address."))

    contact_email_domain = Address(addr_spec=contact_email).domain.lower()
    if contact_email_domain == "example.com":
        raise ServerAdminEmailError(_("example.com is not a valid email domain."))

    # Check if the domain has an MX record
    resolver = dns_resolver.Resolver()
    resolver.timeout = 3
    dns_mx_check_successful = False
    try:
        if resolver.resolve(contact_email_domain, "MX"):
            dns_mx_check_successful = True
    except DNSException:
        pass
    if not dns_mx_check_successful:
        # Check if the A/AAAA exist, for better error reporting
        try:
            resolver.resolve_name(contact_email_domain)
            raise ServerAdminEmailError(
                _("{domain} is invalid because it does not have any MX records").format(
                    domain=contact_email_domain
                )
            )
        except DNSException:
            try:
                resolver.resolve(contact_email_domain, rdtype="NS")
                raise ServerAdminEmailError(
                    _("{domain} is invalid because it does not have any MX records").format(
                        domain=contact_email_domain
                    )
                )
            except DNSException:
                raise ServerAdminEmailError(
                    _("{domain} does not exist").format(domain=contact_email_domain)
                )

    try:
        validate_uuid(zulip_org_id)
    except ValidationError as e:
        raise JsonableError(e.message)

    try:
        remote_server = RemoteZulipServer.objects.get(uuid=zulip_org_id)
    except RemoteZulipServer.DoesNotExist:
        remote_server = None

    if remote_server is not None:
        if not constant_time_compare(remote_server.api_key, zulip_org_key):
            raise InvalidZulipServerKeyError(zulip_org_id)

        if remote_server.deactivated:
            raise RemoteServerDeactivatedError

    if (
        remote_server is None
        and RemoteZulipServer.objects.filter(hostname=hostname, deactivated=False).exists()
    ):
        raise HostnameAlreadyInUseBouncerError(hostname)

    with transaction.atomic(durable=True):
        if remote_server is None:
            created = True
            remote_server = RemoteZulipServer.objects.create(
                uuid=zulip_org_id,
                hostname=hostname,
                contact_email=contact_email,
                api_key=zulip_org_key,
                last_request_datetime=timezone_now(),
            )
            RemoteZulipServerAuditLog.objects.create(
                event_type=AuditLogEventType.REMOTE_SERVER_CREATED,
                server=remote_server,
                event_time=remote_server.last_updated,
            )
        else:
            created = False
            remote_server.hostname = hostname
            remote_server.contact_email = contact_email
            if new_org_key is not None:
                remote_server.api_key = new_org_key

            remote_server.last_request_datetime = timezone_now()
            remote_server.save()

    return json_success(request, data={"created": created})


class RegistrationTransferVerificationSession(OutgoingSession):
    def __init__(self) -> None:
        # The generous timeout and retries here are likely to be unnecessary; a functional Zulip server should
        # respond instantly.
        super().__init__(role="verify_registration_transfer_challenge", timeout=5, max_retries=3)


class EndpointUsageRateLimitError(JsonableError):
    code = ErrorCode.RATE_LIMIT_HIT
    http_status_code = 429


@csrf_exempt
@typed_endpoint
def verify_registration_transfer_challenge_ack_endpoint(
    request: HttpRequest,
    *,
    hostname: str,
    access_token: str,
) -> HttpResponse:
    """
    The host should POST to this endpoint to announce it is ready to serve the received
    secret at {hostname}/zulip-services/verify/{access_token}.
    The access_token is randomly generated by the host in order to prevent 3rd parties
    from accessing the verification secret served at that URL.

    If we successfully verify the secret, we will send the registration credentials
    to the host, completing the whole flow.
    """

    try:
        # This endpoint is at risk of being used to spam another server with our requests,
        # or to freeze up our Django processes by making them wait for timeouts on the
        # requests triggered here.
        # Since this is an extremely low-traffic endpoint, we just put an absolute limit on
        # how many times it can be called in a given time period. There's little value for an
        # attacker to fill up the bucket here, and issues can be handled adequately by
        # manual intervention.
        if settings.RATE_LIMITING:
            rate_limit_endpoint_absolute("verify_registration_transfer_challenge_ack_endpoint")
    except RateLimitedError:
        # This rate limit being hit means we've either set the limits too low for legitimate use,
        # or the endpoint is being spammed. Ideally, we want this endpoint to always be operational
        # so this deserves logging a warning.
        logger.warning(
            "Rate limit exceeded for verify_registration_transfer_challenge_ack_endpoint"
        )
        raise EndpointUsageRateLimitError(
            _(
                "The global limits on recent usage of this endpoint have been reached."
                " Please try again later or reach out to {support_email} for assistance."
            ).format(support_email=FromAddress.SUPPORT)
        )

    try:
        remote_server = RemoteZulipServer.objects.get(hostname=hostname, deactivated=False)
    except RemoteZulipServer.DoesNotExist:
        raise JsonableError(_("Registration not found for this hostname"))

    session = RegistrationTransferVerificationSession()
    url = urljoin(f"https://{hostname}", f"/api/v1/zulip-services/verify/{access_token}/")

    exception_and_error_message: tuple[Exception, str] | None = None
    try:
        response = session.get(url)
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        if check_transfer_challenge_response_secret_not_prepared(e.response):
            logger.info("verify_registration_transfer:host:%s|secret_not_prepared", hostname)
            raise JsonableError(_("The host reported it has no verification secret."))

        error_message = _("Error response received from the host: {status_code}").format(
            status_code=response.status_code
        )
        exception_and_error_message = (e, error_message)
    except requests.exceptions.SSLError as e:
        error_message = "SSL error occurred while communicating with the host."
        exception_and_error_message = (e, error_message)
    except requests.exceptions.ConnectionError as e:
        error_message = "Connection error occurred while communicating with the host."
        exception_and_error_message = (e, error_message)
    except requests.exceptions.Timeout as e:
        error_message = "The request timed out while communicating with the host."
        exception_and_error_message = (e, error_message)
    except requests.exceptions.RequestException as e:
        error_message = "An error occurred while communicating with the host."
        exception_and_error_message = (e, error_message)

    if exception_and_error_message is not None:
        exception, error_message = exception_and_error_message
        logger.info("verify_registration_transfer:host:%s|exception:%s", hostname, exception)
        raise JsonableError(error_message)

    data = response.json()
    verification_secret = data["verification_secret"]
    validate_registration_transfer_verification_secret(verification_secret, hostname)

    logger.info("verify_registration_transfer:host:%s|success", hostname)
    new_secret_key = get_random_string(RemoteZulipServer.API_KEY_LENGTH)
    with transaction.atomic(durable=True):
        remote_server.api_key = new_secret_key
        remote_server.save(update_fields=["api_key"])

        RemoteZulipServerAuditLog.objects.create(
            event_type=AuditLogEventType.REMOTE_SERVER_REGISTRATION_TRANSFERRED,
            server=remote_server,
            event_time=timezone_now(),
        )

    return json_success(
        request,
        data={"zulip_org_id": str(remote_server.uuid), "zulip_org_key": new_secret_key},
    )


def check_transfer_challenge_response_secret_not_prepared(response: requests.Response) -> bool:
    secret_not_prepared = False
    try:
        secret_not_prepared = (
            response.status_code == 400
            and response.json()["code"] == "REMOTE_SERVER_VERIFICATION_SECRET_NOT_PREPARED"
        )
    except Exception:  # nocoverage
        return False
    return secret_not_prepared


def get_remote_push_device_token(
    *,
    server: RemoteZulipServer,
    token: str,
    kind: int,
) -> QuerySet[RemotePushDeviceToken]:
    if kind == RemotePushDeviceToken.APNS:
        return RemotePushDeviceToken.objects.alias(lower_token=Lower("token")).filter(
            server=server,
            lower_token=token.lower(),
            kind=kind,
        )
    else:
        return RemotePushDeviceToken.objects.filter(
            server=server,
            token=token,
            kind=kind,
        )


@typed_endpoint
def register_remote_push_device(
    request: HttpRequest,
    server: RemoteZulipServer,
    *,
    user_id: Json[int] | None = None,
    user_uuid: str | None = None,
    realm_uuid: str | None = None,
    token: Annotated[str, RequiredStringConstraint],
    token_kind: Json[int],
    ios_app_id: ApnsAppId | None = None,
) -> HttpResponse:
    validate_bouncer_token_request(token, token_kind)
    if token_kind == RemotePushDeviceToken.APNS and ios_app_id is None:
        raise JsonableError(_("Missing ios_app_id"))

    if user_id is None and user_uuid is None:
        raise JsonableError(_("Missing user_id or user_uuid"))
    if user_id is not None and user_uuid is not None:
        kwargs: dict[str, object] = {"user_uuid": user_uuid, "user_id": None}
        # Delete pre-existing user_id registration for this user+device to avoid
        # duplication. Further down, uuid registration will be created.
        get_remote_push_device_token(
            server=server,
            token=token,
            kind=token_kind,
        ).filter(user_id=user_id).delete()
    else:
        # One of these is None, so these kwargs will lead to a proper registration
        # of either user_id or user_uuid type
        kwargs = {"user_id": user_id, "user_uuid": user_uuid}

    if realm_uuid is not None:
        # Servers 8.0+ also send the realm.uuid of the user.
        assert isinstance(user_uuid, str), (
            "Servers new enough to send realm_uuid, should also have user_uuid"
        )
        remote_realm = get_remote_realm_helper(request, server, realm_uuid)
        if remote_realm is not None:
            # We want to associate the RemotePushDeviceToken with the RemoteRealm.
            kwargs["remote_realm_id"] = remote_realm.id

            remote_realm.last_request_datetime = timezone_now()
            remote_realm.save(update_fields=["last_request_datetime"])

    RemotePushDeviceToken.objects.bulk_create(
        [
            RemotePushDeviceToken(
                server=server,
                kind=token_kind,
                token=token,
                ios_app_id=ios_app_id,
                # last_updated is to be renamed to date_created.
                last_updated=timezone_now(),
                **kwargs,
            ),
        ],
        ignore_conflicts=True,
    )

    return json_success(request)


class PushRegistration(BaseModel):
    token: str
    token_kind: str
    ios_app_id: ApnsAppId | None = None
    timestamp: int

    def is_valid_token(self) -> bool:
        if self.token == "" or len(self.token) > 4096:
            # Invalid token length
            return False

        if self.token_kind == RemotePushDevice.TokenKind.APNS:
            try:
                bytes.fromhex(self.token)
            except ValueError:
                return False
        return True

    @model_validator(mode="after")
    def validate_terms(self) -> "PushRegistration":
        if self.token_kind not in [RemotePushDevice.TokenKind.APNS, RemotePushDevice.TokenKind.FCM]:
            raise ValueError("Invalid token_kind")

        if self.token_kind == RemotePushDevice.TokenKind.APNS and self.ios_app_id is None:
            raise ValueError("Missing ios_app_id")

        if self.token_kind == RemotePushDevice.TokenKind.FCM and self.ios_app_id is not None:
            raise ValueError(
                f"For token_kind={RemotePushDevice.TokenKind.FCM}, ios_app_id should be null"
            )

        if not self.is_valid_token():
            raise ValueError("Invalid token")

        return self


def do_register_remote_push_device(
    bouncer_public_key: str,
    encrypted_push_registration: str,
    push_account_id: int,
    *,
    realm: Realm | None = None,
    remote_realm: RemoteRealm | None = None,
) -> int:
    assert (realm is None) ^ (remote_realm is None)

    assert settings.PUSH_REGISTRATION_ENCRYPTION_KEYS
    if bouncer_public_key not in settings.PUSH_REGISTRATION_ENCRYPTION_KEYS:
        raise InvalidBouncerPublicKeyError

    # Decrypt push_registration
    bouncer_private_key: str = settings.PUSH_REGISTRATION_ENCRYPTION_KEYS[bouncer_public_key]
    private_key = PrivateKey(bouncer_private_key.encode("utf-8"), encoder=Base64Encoder)
    unseal_box = SealedBox(private_key)

    try:
        push_registration_bytes = unseal_box.decrypt(
            Base64Encoder.decode(encrypted_push_registration.encode("utf-8"))
        )
    except (TypeError, CryptoError):
        raise InvalidEncryptedPushRegistrationError

    try:
        push_registration = PushRegistration.model_validate_json(push_registration_bytes)
    except PydanticValidationError:
        raise InvalidEncryptedPushRegistrationError

    if (
        datetime_to_timestamp(timezone_now()) - push_registration.timestamp
        > PUSH_REGISTRATION_LIVENESS_TIMEOUT
    ):
        raise RequestExpiredError

    # If already registered, return the device_id.
    # The query uses the unique index created by the
    # 'unique_remote_push_device_push_account_id_token' constraint.
    remote_push_device = RemotePushDevice.objects.filter(
        token=push_registration.token, push_account_id=push_account_id
    ).first()
    if remote_push_device:
        return remote_push_device.device_id

    remote_push_device = RemotePushDevice.objects.create(
        realm=realm,
        remote_realm=remote_realm,
        token=push_registration.token,
        token_kind=push_registration.token_kind,
        push_account_id=push_account_id,
        ios_app_id=push_registration.ios_app_id,
    )
    return remote_push_device.device_id


@typed_endpoint
def register_remote_push_device_for_e2ee_push_notification(
    request: HttpRequest,
    server: RemoteZulipServer,
    *,
    realm_uuid: str,
    push_account_id: Json[int],
    encrypted_push_registration: str,
    bouncer_public_key: str,
) -> HttpResponse:
    remote_realm = get_remote_realm_helper(request, server, realm_uuid)
    if remote_realm is None:
        raise MissingRemoteRealmError
    else:
        remote_realm.last_request_datetime = timezone_now()
        remote_realm.save(update_fields=["last_request_datetime"])

    device_id = do_register_remote_push_device(
        bouncer_public_key,
        encrypted_push_registration,
        push_account_id,
        remote_realm=remote_realm,
    )

    return json_success(request, {"device_id": device_id})


@typed_endpoint
def unregister_remote_push_device(
    request: HttpRequest,
    server: RemoteZulipServer,
    *,
    token: Annotated[str, RequiredStringConstraint],
    token_kind: Json[int],
    user_id: Json[int] | None = None,
    user_uuid: str | None = None,
    realm_uuid: str | None = None,
) -> HttpResponse:
    validate_bouncer_token_request(token, token_kind)
    user_identity = UserPushIdentityCompat(user_id=user_id, user_uuid=user_uuid)

    update_remote_realm_last_request_datetime_helper(request, server, realm_uuid, user_uuid)

    (num_deleted, _deletions) = (
        get_remote_push_device_token(token=token, kind=token_kind, server=server)
        .filter(user_identity.filter_q())
        .delete()
    )
    if num_deleted == 0:
        raise JsonableError(err_("Token does not exist"))

    return json_success(request)


@typed_endpoint
def unregister_all_remote_push_devices(
    request: HttpRequest,
    server: RemoteZulipServer,
    *,
    user_id: Json[int] | None = None,
    user_uuid: str | None = None,
    realm_uuid: str | None = None,
) -> HttpResponse:
    user_identity = UserPushIdentityCompat(user_id=user_id, user_uuid=user_uuid)

    update_remote_realm_last_request_datetime_helper(request, server, realm_uuid, user_uuid)

    RemotePushDeviceToken.objects.filter(user_identity.filter_q(), server=server).delete()
    return json_success(request)


def update_remote_realm_last_request_datetime_helper(
    request: HttpRequest,
    server: RemoteZulipServer,
    realm_uuid: str | None,
    user_uuid: str | None,
) -> None:
    if realm_uuid is not None:
        assert user_uuid is not None
        remote_realm = get_remote_realm_helper(request, server, realm_uuid)
        if remote_realm is not None:
            remote_realm.last_request_datetime = timezone_now()
            remote_realm.save(update_fields=["last_request_datetime"])


def delete_duplicate_registrations(
    registrations: list[RemotePushDeviceToken], server_id: int, user_id: int, user_uuid: str
) -> list[RemotePushDeviceToken]:
    """
    When migrating to support registration by UUID, we introduced a bug where duplicate
    registrations for the same device+user could be created - one by user_id and one by
    user_uuid. Given no good way of detecting these duplicates at database level, we need to
    take advantage of the fact that when a remote server sends a push notification request
    to us, it sends both user_id and user_uuid of the user.
    See https://github.com/zulip/zulip/issues/24969 for reference.

    This function, knowing the user_id and user_uuid of the user, can detect duplicates
    and delete the legacy user_id registration if appropriate.

    Return the list of registrations with the user_id-based duplicates removed.
    """

    # All registrations passed here should be of the same kind (apple vs android).
    assert len({registration.kind for registration in registrations}) == 1
    kind = registrations[0].kind

    if kind == RemotePushDeviceToken.APNS:
        tokens_counter = Counter(device.token.lower() for device in registrations)
    else:
        tokens_counter = Counter(device.token for device in registrations)

    tokens_to_deduplicate = []
    for key in tokens_counter:
        if tokens_counter[key] <= 1:
            continue
        if tokens_counter[key] > 2:
            raise AssertionError(
                f"More than two registrations for token {key} for user id:{user_id} uuid:{user_uuid}, shouldn't be possible"
            )
        assert tokens_counter[key] == 2
        tokens_to_deduplicate.append(key)

    if not tokens_to_deduplicate:
        return registrations

    logger.info(
        "Deduplicating push registrations for server id:%s user id:%s uuid:%s and tokens:%s",
        server_id,
        user_id,
        user_uuid,
        sorted(tokens_to_deduplicate),
    )
    RemotePushDeviceToken.objects.filter(
        token__in=tokens_to_deduplicate, kind=kind, server_id=server_id, user_id=user_id
    ).delete()

    deduplicated_registrations_to_return = []
    for registration in registrations:
        if registration.token in tokens_to_deduplicate and registration.user_id is not None:
            # user_id registrations are the ones we deleted
            continue
        deduplicated_registrations_to_return.append(registration)

    return deduplicated_registrations_to_return


class TestNotificationPayload(BaseModel):
    token: str
    token_kind: int
    user_id: int
    user_uuid: str
    realm_uuid: str | None = None
    base_payload: dict[str, Any]

    model_config = ConfigDict(extra="forbid")


@typed_endpoint
def remote_server_send_test_notification(
    request: HttpRequest,
    server: RemoteZulipServer,
    *,
    payload: JsonBodyPayload[TestNotificationPayload],
) -> HttpResponse:
    token = payload.token
    token_kind = payload.token_kind

    user_id = payload.user_id
    user_uuid = payload.user_uuid
    realm_uuid = payload.realm_uuid

    # The remote server only sends the base payload with basic user and server info,
    # and the actual format of the test notification is defined on the bouncer, as that
    # gives us the flexibility to modify it freely, without relying on other servers
    # upgrading.
    base_payload = payload.base_payload

    # This is a new endpoint, so it can assume it will only be used by newer
    # servers that will send user both UUID and ID.
    user_identity = UserPushIdentityCompat(user_id=user_id, user_uuid=user_uuid)

    update_remote_realm_last_request_datetime_helper(request, server, realm_uuid, user_uuid)

    device = (
        get_remote_push_device_token(token=token, kind=token_kind, server=server)
        .filter(user_identity.filter_q())
        .first()
    )
    if device is None:
        raise InvalidRemotePushDeviceTokenError

    send_test_push_notification_directly_to_devices(
        user_identity, [device], base_payload, remote=server
    )
    return json_success(request)


def get_remote_realm_helper(
    request: HttpRequest,
    server: RemoteZulipServer,
    realm_uuid: str,
) -> RemoteRealm | None:
    """
    Tries to fetch RemoteRealm for the given realm_uuid and server. Otherwise,
    returns None and logs what happened using request and user_uuid args to make
    the output more informative.
    """

    try:
        remote_realm = RemoteRealm.objects.get(uuid=realm_uuid)
    except RemoteRealm.DoesNotExist:
        logger.info(
            "%s: Received request for unknown realm %s, server %s",
            request.path,
            realm_uuid,
            server.id,
        )
        return None

    if remote_realm.server_id != server.id:
        logger.warning(
            "%s: Realm %s exists, but not registered to server %s",
            request.path,
            realm_uuid,
            server.id,
        )
        raise RemoteRealmServerMismatchError

    return remote_realm


class OldZulipServerError(JsonableError):
    code = ErrorCode.INVALID_ZULIP_SERVER

    def __init__(self, msg: str) -> None:
        self._msg: str = msg


class PushNotificationsDisallowedError(JsonableError):
    code = ErrorCode.PUSH_NOTIFICATIONS_DISALLOWED

    def __init__(self, reason: str) -> None:
        msg = _(
            "Your plan doesn't allow sending push notifications. Reason provided by the server: {reason}"
        ).format(reason=reason)
        super().__init__(msg)


class RemoteServerNotificationPayload(BaseModel):
    user_id: int | None = None
    user_uuid: str | None = None
    realm_uuid: str | None = None
    gcm_payload: dict[str, Any] = {}
    apns_payload: dict[str, Any] = {}
    gcm_options: dict[str, Any] = {}

    android_devices: list[str] = []
    apple_devices: list[str] = []


@typed_endpoint
def remote_server_notify_push(
    request: HttpRequest,
    server: RemoteZulipServer,
    *,
    payload: JsonBodyPayload[RemoteServerNotificationPayload],
) -> HttpResponse:
    from corporate.lib.stripe import get_push_status_for_remote_request

    user_id = payload.user_id
    user_uuid = payload.user_uuid
