FCM push: ilovasi yopiq foydalanuvchiga xabar yetkazish

Alert saqlandi (post_save signal)
        │
        ├──► Channels group_send ──► ochiq ilovalar (2-qism)
        │
        └──► FCM API ──► Google/Apple serverlari ──► yopiq ilovalar (shu qism)

uv add firebase-admin

# config/settings.py
import os

# kalit yo'li .env orqali beriladi: FIREBASE_CREDENTIALS=/opt/app/secrets/firebase.json
FIREBASE_CREDENTIALS = os.environ.get("FIREBASE_CREDENTIALS", "")

# apps/notifications/firebase.py
import firebase_admin
from django.conf import settings
from firebase_admin import credentials


def init_firebase() -> None:
    # ikki marta chaqirilsa ham xato bermasin — idempotent
    if firebase_admin._apps:
        return
    cred = credentials.Certificate(settings.FIREBASE_CREDENTIALS)
    firebase_admin.initialize_app(cred)

apps/notifications/
├── __init__.py
├── firebase.py      ← SDK init (yuqorida)
├── models.py        ← UserDevice
├── serializers.py
├── views.py         ← token ro'yxatga olish endpoint'i
├── services.py      ← push yuborish logikasi
└── tests.py

# apps/notifications/models.py
from django.conf import settings
from django.db import models


class UserDevice(models.Model):
    class Platform(models.TextChoices):
        ANDROID = "android", "Android"
        IOS = "ios", "iOS"

    # null=True — mehmon-qurilma: hali login qilmagan, lekin push oladi.
    # Keyin login qilsa, xuddi shu yozuvga user bog'lanadi.
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        on_delete=models.CASCADE,
        related_name="devices",
    )
    token = models.CharField(max_length=512, unique=True)
    platform = models.CharField(max_length=8, choices=Platform.choices)
    is_active = models.BooleanField(default=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return f"{self.platform} · {self.token[:12]}…"

# apps/notifications/serializers.py
from rest_framework import serializers

from .models import UserDevice


class DeviceSerializer(serializers.ModelSerializer):
    class Meta:
        model = UserDevice
        fields = ["token", "platform"]

# apps/notifications/views.py
from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework.views import APIView

from .models import UserDevice
from .serializers import DeviceSerializer


class RegisterDeviceView(APIView):
    # mehmonlar ham ro'yxatga olinadi — mahsulot qarori (yuqoriga qarang)
    permission_classes = [AllowAny]

    def post(self, request):
        serializer = DeviceSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        device, _ = UserDevice.objects.update_or_create(
            token=serializer.validated_data["token"],
            defaults={
                "platform": serializer.validated_data["platform"],
                # login bo'lgan bo'lsa qurilma egasini bog'laymiz,
                # mehmon bo'lsa None qoladi
                "user": request.user if request.user.is_authenticated else None,
                "is_active": True,
            },
        )
        return Response({"ok": True}, status=status.HTTP_201_CREATED)

# apps/notifications/services.py
from firebase_admin import messaging

from .firebase import init_firebase
from .models import UserDevice

BATCH = 500  # FCM cheklovi: bitta multicast'da 500 tagacha token


def send_push(title: str, body: str, data: dict[str, str] | None = None) -> dict:
    init_firebase()
    tokens = list(UserDevice.objects.filter(is_active=True).values_list("token", flat=True))
    stats = {"sent": 0, "dead": 0}

    for start in range(0, len(tokens), BATCH):
        chunk = tokens[start : start + BATCH]
        message = messaging.MulticastMessage(
            tokens=chunk,
            notification=messaging.Notification(title=title, body=body),
            # data — ilovaga "ko'rinmas" yuk: masalan qaysi ekranni ochish
            data=data or {},
        )
        response = messaging.send_each_for_multicast(message)
        stats["sent"] += response.success_count

        # javob tartibi token tartibiga mos — o'liklarni aniqlaymiz
        dead = [
            chunk[i]
            for i, result in enumerate(response.responses)
            if result.exception is not None
            and isinstance(result.exception, messaging.UnregisteredError)
        ]
        if dead:
            # qattiq o'chirmaymiz — flag: qurilma qaytib kelsa qayta yoqiladi
            UserDevice.objects.filter(token__in=dead).update(is_active=False)
            stats["dead"] += len(dead)

    return stats

# apps/alerts/signals.py  (2-qismdagi faylning davomi)
import threading

from apps.notifications.services import send_push


@receiver(post_save, sender=Alert)
def push_alert(sender, instance: Alert, created: bool, **kwargs):
    if not created or not instance.is_active:
        return
    # FCM chaqiruvi soniyalar olishi mumkin — admin so'rovini kutdirmaslik
    # uchun alohida oqimda. Yuk oshsa bu joy Celery vazifasiga ko'chadi:
    # https://solijonov.uz/writing/celery-fon-vazifalar/
    threading.Thread(
        target=send_push,
        args=(instance.title, instance.body),
        kwargs={"data": {"alert_id": str(instance.id)}},
        daemon=True,
    ).start()

# apps/notifications/tests.py
from unittest.mock import MagicMock, patch

import pytest
from firebase_admin import messaging

from apps.notifications.models import UserDevice
from apps.notifications.services import send_push


@pytest.fixture
def devices(db):
    UserDevice.objects.create(token="tok-tirik", platform="android")
    UserDevice.objects.create(token="tok-olik", platform="ios")


def _fake_response():
    ok = MagicMock(exception=None)
    dead = MagicMock(exception=messaging.UnregisteredError("gone"))
    return MagicMock(success_count=1, responses=[ok, dead])


@patch("apps.notifications.services.init_firebase")
@patch("apps.notifications.services.messaging.send_each_for_multicast")
def test_dead_tokens_get_deactivated(send_mock, init_mock, devices):
    send_mock.return_value = _fake_response()

    stats = send_push("Sinov", "Matn")

    assert stats == {"sent": 1, "dead": 1}
    assert UserDevice.objects.get(token="tok-olik").is_active is False
    assert UserDevice.objects.get(token="tok-tirik").is_active is True