zulip/zerver/models/push_notifications.py

from typing import Literal

from django.db import models
from django.db.models import CASCADE, F, Q, UniqueConstraint
from django.db.models.functions import Lower

from zerver.lib.exceptions import (
    InvalidBouncerPublicKeyError,
    InvalidEncryptedPushRegistrationError,
    RequestExpiredError,
)
from zerver.models.users import UserProfile


class AbstractPushDeviceToken(models.Model):
    APNS = 1
    FCM = 2

    KINDS = (
        (APNS, "apns"),
        # The string value in the database is "gcm" for legacy reasons.
        # TODO: We should migrate it.
        (FCM, "gcm"),
    )

    kind = models.PositiveSmallIntegerField(choices=KINDS)

    # The token is a unique device-specific token that is
    # sent to us from each device:
    #   - APNS token if kind == APNS
    #   - FCM registration id if kind == FCM
    token = models.CharField(max_length=4096, db_index=True)

    # TODO: last_updated should be renamed date_created, since it is
    # no longer maintained as a last_updated value.
    last_updated = models.DateTimeField(auto_now=True)

    # [optional] Contains the app id of the device if it is an iOS device
    ios_app_id = models.TextField(null=True)

    class Meta:
        abstract = True


class PushDeviceToken(AbstractPushDeviceToken):
    # The user whose device this is
    user = models.ForeignKey(UserProfile, db_index=True, on_delete=CASCADE)

    class Meta:
        constraints = [
            models.UniqueConstraint(
                "user_id",
                "kind",
                Lower(F("token")),
                name="zerver_pushdevicetoken_apns_user_kind_token",
                condition=Q(kind=AbstractPushDeviceToken.APNS),
            ),
            models.UniqueConstraint(
                "user_id",
                "kind",
                "token",
                name="zerver_pushdevicetoken_fcm_user_kind_token",
                condition=Q(kind=AbstractPushDeviceToken.FCM),
            ),
        ]


class AbstractPushDevice(models.Model):
    class TokenKind(models.TextChoices):
        APNS = "apns", "APNs"
        FCM = "fcm", "FCM"

    token_kind = models.CharField(max_length=4, choices=TokenKind.choices)

    # 64-bit random integer ID generated by the client; will only be
    # guaranteed to be unique within the client's own table of accounts.
    push_account_id = models.BigIntegerField()

    class Meta:
        abstract = True


class PushDevice(AbstractPushDevice):
    """Core zulip server table storing registrations that are potentially
    registered with the mobile push notifications bouncer service.

    Each row corresponds to an account on an install of the app
    that has attempted to register with the bouncer to receive
    mobile push notifications.
    """

    # The user on this server to whom this PushDevice belongs.
    user = models.ForeignKey(UserProfile, on_delete=CASCADE)

    # Key to use to encrypt notifications for delivery to this device.
    push_public_key = models.TextField()

    # `device_id` of the corresponding `RemotePushDevice`
    # row created after successful registration to bouncer.
    # Set to NULL while registration to bouncer is in progress or failed.
    bouncer_device_id = models.BigIntegerField(null=True)

    class ErrorCode(models.TextChoices):
        INVALID_BOUNCER_PUBLIC_KEY = InvalidBouncerPublicKeyError.code.name
        INVALID_ENCRYPTED_PUSH_REGISTRATION = InvalidEncryptedPushRegistrationError.code.name
        REQUEST_EXPIRED = RequestExpiredError.code.name

    # The error code returned when registration to bouncer fails.
    # Set to NULL if the registration is in progress or successful.
    error_code = models.CharField(max_length=100, choices=ErrorCode.choices, null=True)

    class Meta:
        constraints = [
            UniqueConstraint(
                # In theory, there's possibility that a user with
                # two different devices having the same 'push_account_id'
                # generated by the client to register. But the API treats
                # that as idempotent request.
                # We treat (user, push_account_id) as a unique registration.
                #
                # Also, the unique index created is used by queries in `get_push_accounts`,
                # `register_push_device`, and `handle_register_push_device_to_bouncer`.
                fields=["user", "push_account_id"],
                name="unique_push_device_user_push_account_id",
            ),
        ]
        indexes = [
            models.Index(
                # Used in 'send_push_notifications' function,
                # in 'zerver/lib/push_notifications'.
                fields=["user", "bouncer_device_id"],
                condition=Q(bouncer_device_id__isnull=False),
                name="zerver_pushdevice_user_bouncer_device_id_idx",
            ),
        ]

    @property
    def status(self) -> Literal["active", "pending", "failed"]:
        if self.error_code is not None:
            return "failed"
        elif self.bouncer_device_id is None:
            return "pending"
        return "active"