Add campaigns/ Add supervisor/ Add messaging features/ Separate SMS

campaigns/services.py

@@ -0,0 +1,244 @@
+"""
+اجرای یک MessageCampaign: صرفاً «چسباندنِ» سه چیزِ از قبل موجود به هم —
+هدف‌گیریِ کاربر (accounts.services)، ارسالِ پیامک (accounts.services.send_sms)
+و ثبتِ اعلانِ داخل‌اپ (notifications.services.notify). این ماژول هیچ منطقِ
+جدیدِ ارسال/اعلان نمی‌سازه، هیچ صف/جاب پس‌زمینه‌ای نداره، و هیچ زمان‌بندی،
+retry یا پیگیریِ تحویلی انجام نمی‌ده — طبق MessageCampaign.__doc__، اونا
+مسئولیتِ فیچرهای بعدی هستن.
+
+فراخوانی همیشه sync و بلاکینگ‌ه (برای هر کاربر یک send_sms/notify)؛
+مناسبِ لیست‌های کوچیک یا اجرای دستی/تست، نه کمپین‌های بزرگ در پروداکشن.
+"""
+from django.db import IntegrityError, transaction
+
+from accounts import services as account_services
+from accounts.services import SMSDeliveryError
+from notifications.services import notify
+
+from .models import CampaignDelivery, MessageCampaign
+
+# نوعِ نوتیفیکیشنِ ثبت‌شده برای اعلان‌های ناشی از کمپین. عمداً در
+# notifications.models.Notification.TYPE_CHOICES اضافه نشده چون آن
+# choices فقط رویدادهایی‌ان که notifications به‌عنوانِ event-listener
+# بهشون گوش می‌ده (طبق داکیومنتِ همون فایل)؛ این‌جا notify() مستقیماً و
+# بدونِ رویداد صدا زده می‌شه، پس تغییرِ آن مدل لازم نیست — Notification.type
+# یک CharField معمولیه و مقداردهی‌اش خارج از choices هم مجاز است.
+CAMPAIGN_NOTIFICATION_TYPE = 'message_campaign'
+
+
+class UnsupportedTargetFilter(ValueError):
+    """target_filter کمپین با هیچ‌کدوم از حالت‌های فعلاً پشتیبانی‌شده مطابقت نداره."""
+
+
+def _resolve_target_users(target_filter: dict):
+    """
+    target_filter رو به یک QuerySet از accounts.User تبدیل می‌کنه، فقط با
+    استفاده از هلپرهای موجودِ accounts.services (بدونِ کوئریِ جدید).
+
+    حالت‌های فعلاً پشتیبانی‌شده:
+        {"type": "all"}
+        {"type": "grade", "grade_id": <int>}
+        {"type": "inactive_since_days", "days": <int>}
+        {"type": "active_subscription"}
+    """
+    target_filter = target_filter or {}
+    filter_type = target_filter.get('type')
+
+    if filter_type == 'all':
+        return account_services.get_all_users()
+
+    if filter_type == 'grade':
+        return account_services.get_users_by_grade(target_filter['grade_id'])
+
+    if filter_type == 'inactive_since_days':
+        from datetime import timedelta
+
+        from django.utils import timezone
+
+        since = timezone.now() - timedelta(days=target_filter['days'])
+        return account_services.get_inactive_users(since)
+
+    if filter_type == 'active_subscription':
+        return account_services.get_users_with_active_subscription()
+
+    raise UnsupportedTargetFilter(
+        f"Unsupported target_filter type: {filter_type!r}"
+    )
+
+
+def _mark_delivery_result(delivery: CampaignDelivery, *, error: Exception | None) -> None:
+    """رکوردِ pending‌ای که قبل از تلاشِ ارسال ساخته شده رو به sent/failed آپدیت می‌کنه."""
+    if error is None:
+        delivery.status = CampaignDelivery.STATUS_SENT
+        delivery.save(update_fields=['status'])
+        return
+
+    delivery.status = CampaignDelivery.STATUS_FAILED
+    delivery.error_message = str(error)[:300]
+    delivery.save(update_fields=['status', 'error_message'])
+
+
+def _claim_delivery(campaign: MessageCampaign, user_id: int) -> CampaignDelivery | None:
+    """
+    رکوردِ pendingِ یک CampaignDelivery رو برای این کاربر «claim» می‌کنه،
+    یا اگه یک اجرای هم‌زمانِ دیگرِ همین campaign همین کاربر رو زودتر
+    claim کرده باشه، None برمی‌گردونه (بدونِ ساختنِ رکورد و بدونِ
+    فراخوانیِ send_sms/notify).
+
+    چکِ already_sent_user_ids قبل از شروعِ حلقه یک‌بار انجام می‌شه، پس
+    بینِ اون چک و رسیدنِ نوبتِ این کاربر در حلقه، ممکنه یک اجرای هم‌زمانِ
+    دیگر (پراسسِ دیگر/دابل‌کلیکِ اجرا) دقیقاً همین کاربر رو claim کرده و
+    حتی به sent رسونده باشه (TOCTOU). این تابع همون کاری‌ه که چکِ
+    اپلیکیشن‌لایه فرض می‌کرد ولی گارانتی نمی‌داد: با تکیه به
+    CampaignDelivery.Meta.constraints (یک رکوردِ فعال/موفق به ازای هر
+    campaign+user_id+channel)، اگه یک رکوردِ pending یا sent از قبل
+    برای همین کاربر وجود داشته باشه (چه توسطِ این اجرا چه توسطِ اجرای
+    هم‌زمانِ دیگر)، create() با IntegrityError رد می‌شه و این تابع
+    None برمی‌گردونه — یعنی کاربر به‌جای duplicate delivery، skip می‌شه.
+    """
+    try:
+        with transaction.atomic():
+            return CampaignDelivery.objects.create(
+                campaign=campaign, user_id=user_id, channel=campaign.channel,
+            )
+    except IntegrityError:
+        return None
+
+
+def _already_sent_user_ids(campaign: MessageCampaign) -> set[int]:
+    """
+    user_id‌هایی که برای همین campaign+channel از قبل یک CampaignDelivery
+    با status=sent دارن — یعنی تحویلِ واقعی قبلاً با موفقیت انجام شده.
+
+    این استعلام دقیقاً همون کاری‌ه که پرامپت به‌عنوانِ جهتِ حل پیشنهاد
+    داده: «استفاده از رکوردهای CampaignDelivery موجود برای جلوگیری از
+    ارسالِ دوباره‌ی تحویل‌هایی که قبلاً کامل شدن». عمداً روی pending/failed
+    فیلتر نمی‌کنه: pending یعنی تلاشِ قبلی هیچ‌وقت به نتیجه نرسیده (مثلاً
+    اجرا وسط کار قطع شده) و failed یعنی واقعاً تحویل داده نشده — در هر دو
+    حالت، تلاشِ دوباره «duplicate delivery» نیست، بلکه اولین تحویلِ واقعیِ
+    موفقه؛ پس این کاربرها باید دوباره امتحان بشن.
+    """
+    return set(
+        CampaignDelivery.objects.filter(
+            campaign=campaign,
+            channel=campaign.channel,
+            status=CampaignDelivery.STATUS_SENT,
+        ).values_list('user_id', flat=True)
+    )
+
+
+def execute_campaign(campaign: MessageCampaign) -> None:
+    """
+    یک MessageCampaign رو اجرا می‌کنه: کاربرانِ هدف رو resolve می‌کنه و به
+    ازای هر کدوم، بسته به campaign.channel، یا اعلانِ داخل‌اپ ثبت می‌کنه
+    (notify) یا پیامک می‌فرسته (send_sms). به ازای هر تلاشِ واقعیِ ارسال،
+    یک CampaignDelivery ساخته می‌شه (ابتدا pending) و بعد از تلاش به
+    sent/failed آپدیت می‌شه — بدونِ retry، صفِ پس‌زمینه یا زمان‌بندی.
+
+    Idempotency: اگه execute_campaign() دوباره برای همون campaign صدا زده
+    بشه (مثلاً به‌خاطرِ اجرای دستیِ تکراری یا ری‌استارتِ فرآیند وسطِ کار)،
+    کاربرهایی که از قبل یک CampaignDelivery با status=sent برای همین
+    campaign+channel دارن، دوباره skip می‌شن — نه CampaignDelivery‌ی
+    جدیدی براشون ساخته می‌شه و نه send_sms/notify دوباره صداشون می‌زنه.
+    این چک قبل از شروعِ حلقه، یک‌بار برای کلِ کمپین انجام می‌شه (نه به
+    ازای هر کاربر) تا کوئریِ اضافه نداشته باشیم. کاربرهایی که رکوردشون
+    pending یا failed مونده، عمداً دوباره امتحان می‌شن؛ چون براشون هنوز
+    هیچ تحویلِ موفقی ثبت نشده.
+
+    Concurrency: چکِ بالا خودش TOCTOU‌ه (یک‌بار قبل از حلقه محاسبه
+    می‌شه)، پس اگه دو اجرای execute_campaign() برای همون campaign
+    هم‌زمان راه بیفتن، ممکنه هر دو یک کاربر رو «هنوز sent نیست» ببینن.
+    محافظتِ واقعی در _claim_delivery() و در سطحِ دیتابیسه: هر کاربر قبل
+    از notify/send_sms باید یک رکوردِ CampaignDelivery رو claim کنه، و
+    یک unique constraint روی مدل (به ازای هر campaign+user_id+channel
+    حداکثر یک رکوردِ pending/sent) تضمین می‌کنه که فقط یکی از اون دو
+    اجرای هم‌زمان می‌تونه claim موفق داشته باشه؛ اون یکی دیگه بدونِ
+    ساختنِ رکورد و بدونِ صدازدنِ notify/send_sms، آن کاربر رو skip
+    می‌کنه.
+
+    کاربرانِ مهمان بدونِ شماره‌تلفن، برای کانالِ sms به‌سادگی رد (skip)
+    می‌شن و هیچ CampaignDelivery‌ای براشون ساخته نمی‌شه — چون این‌جا
+    اصلاً تلاشی برای ارسال انجام نشده؛ accounts.services.send_sms برای
+    شماره‌ی خالی معنی نداره و هیچ صف/ارسالِ جایگزینی هم براشون نیست.
+
+    شکستِ تحویلِ sms (SMSDeliveryError) دیگه کلِ اجرای کمپین رو متوقف
+    نمی‌کنه؛ فقط همون رکورد failed می‌شه و اجرا برای بقیه‌ی کاربران ادامه
+    پیدا می‌کنه.
+
+    Consistency (مارک‌کردنِ sent بعد از تلاشِ موفق): برای in_app، هم
+    notify() (یک Notification.objects.create ساده) و هم آپدیتِ نهاییِ
+    CampaignDelivery به sent، هر دو نوشتنِ دیتابیسیِ ساده روی همین
+    دیتابیسن — پس هر دو تویِ یک `transaction.atomic()` انجام می‌شن. اگه
+    notify() خودش شکست بخوره یا اگه بعد از notify() موفق، خودِ نوشتنِ
+    «sent» به هر دلیلی (مثلاً یک خطای دیتابیسِ گذرا) شکست بخوره، کل
+    تراکنش rollback می‌شه — یعنی Notification‌ای هم که notify() نوشته
+    بود هیچ‌وقت persist نمی‌شه، و بعدش رکورد به failed آپدیت می‌شه (بیرونِ
+    همون تراکنشِ rollback‌شده، با یک نوشتنِ مستقلِ جدید) و exception
+    دوباره بالا می‌ره. نتیجه: هیچ‌وقت به وضعیتِ ناسازگارِ «Notification
+    واقعاً وجود داره ولی CampaignDelivery رو pendingِ گیرکرده مونده»
+    نمی‌رسیم؛ رکورد یا واقعاً sent می‌شه (با Notificationِ واقعی) یا
+    failed می‌شه (بدونِ هیچ Notificationِ persist‌شده‌ای) — و چون failed
+    از unique constraint مستثناست، برای اجرای بعدی قابلِ retry می‌مونه.
+
+    برای sms این کار عمداً انجام نشده و نمی‌تونه انجام بشه: send_sms()
+    یک تماسِ خارجی با provider‌ه، نه یک نوشتنِ دیتابیسی — یک
+    transaction.atomic() فقط نوشتن‌های همین دیتابیس رو rollback می‌کنه؛
+    پیامکی که provider واقعاً فرستاده رو نمی‌شه با rollbackِ دیتابیس
+    «پس گرفت». اگه بعد از یک send_sms() موفق، خودِ نوشتنِ «sent» شکست
+    بخوره، exception (بدونِ catch/hide) بالا می‌ره و رکورد pending
+    می‌مونه — دقیقاً همون رفتارِ قبلی. این pendingِ گیرکرده باعثِ
+    duplicate SMS در اجرای بعدی نمی‌شه (چون unique constraint یک
+    claimِ دوباره برای همون کاربر رو رد می‌کنه)، ولی اون یک رکوردِ
+    خاص هم دیگه هیچ‌وقت خودکار به sent/failed نمی‌رسه — نیازمندِ
+    بررسیِ دستیه، که پیامدِ ذاتیِ غیرقابل‌اجتنابِ غیرترانزکشنال‌بودنِ
+    provider خارجیه، نه یک باگِ قابل‌رفعِ کوچیک.
+    """
+    users = _resolve_target_users(campaign.target_filter)
+    already_sent_user_ids = _already_sent_user_ids(campaign)
+
+    if campaign.channel == 'in_app':
+        for user in users.only('id').iterator():
+            if user.id in already_sent_user_ids:
+                continue
+            delivery = _claim_delivery(campaign, user.id)
+            if delivery is None:
+                continue
+            try:
+                with transaction.atomic():
+                    notify(
+                        user_id=user.id,
+                        type=CAMPAIGN_NOTIFICATION_TYPE,
+                        title=campaign.title,
+                        body=campaign.body,
+                        related_id=campaign.id,
+                    )
+                    _mark_delivery_result(delivery, error=None)
+            except Exception as exc:
+                # چه notify() خودش شکست بخوره چه فقط نوشتنِ نهاییِ sent
+                # (که تویِ همون atomic block بالاست)، هر دو حالت اینجا
+                # می‌رسه: تراکنش rollback شده (پس هیچ Notificationِ
+                # persist‌شده‌ای نمونده)، و این save یک نوشتنِ جدید و
+                # مستقله که رکورد رو به failed می‌بره تا برای اجرای بعدی
+                # گیر نکنه. exception عمداً catch/hide نمی‌شه و بالا می‌ره.
+                _mark_delivery_result(delivery, error=exc)
+                raise
+        return
+
+    if campaign.channel == 'sms':
+        for user in users.only('id', 'phone').iterator():
+            if user.id in already_sent_user_ids:
+                continue
+            if not user.phone:
+                continue
+            delivery = _claim_delivery(campaign, user.id)
+            if delivery is None:
+                continue
+            try:
+                account_services.send_sms(phone=user.phone, message=campaign.body)
+            except SMSDeliveryError as exc:
+                _mark_delivery_result(delivery, error=exc)
+            else:
+                _mark_delivery_result(delivery, error=None)
+        return
+
+    raise ValueError(f"Unsupported campaign channel: {campaign.channel!r}")