MathApp/campaigns/services.py

"""
اجرای یک MessageCampaign: صرفاً «چسباندنِ» سه چیزِ از قبل موجود به هم —
هدف‌گیریِ کاربر (accounts.services)، ارسالِ پیامک (accounts.services.send_sms)
و ثبتِ اعلانِ داخل‌اپ (notifications.services.notify). این ماژول هیچ منطقِ
جدیدِ ارسال/اعلان نمی‌سازه، هیچ صف/جاب پس‌زمینه‌ای نداره، و هیچ زمان‌بندی،
retry یا پیگیریِ تحویلی انجام نمی‌ده — طبق MessageCampaign.__doc__، اونا
مسئولیتِ فیچرهای بعدی هستن.

فراخوانی همیشه sync و بلاکینگ‌ه (برای هر کاربر یک send_sms/notify)؛
مناسبِ لیست‌های کوچیک یا اجرای دستی/تست، نه کمپین‌های بزرگ در پروداکشن.
"""
import time

from django.db import IntegrityError, OperationalError, transaction

from accounts import services as account_services
from accounts.services import SMSDeliveryError
from notifications.services import notify

from .models import CampaignDelivery, MessageCampaign


def _retry_on_sqlite_lock(fn, *, retries=5, base_delay=0.05):
    """
    روی SQLite (مخصوصاً دیتابیسِ حافظه‌ایِ shared-cache که تست‌ها ازش
    استفاده می‌کنن)، وقتی دو اتصالِ هم‌زمان می‌خوان همون جدول رو بنویسن،
    ممکنه OperationalError('database table is locked') بگیریم. این
    SQLITE_LOCKED‌ه نه SQLITE_BUSY، پس busy_timeoutِ معمولیِ اتصال کمکی
    نمی‌کنه و بلافاصله raise می‌شه؛ یک خطای واقعیِ داده نیست، فقط
    تناقضِ لحظه‌ایِ لاکه. این تابع فقط همین خطای گذرا رو با چند بار
    تلاشِ سریع (backoffِ کوچیکِ فزاینده) دوباره امتحان می‌کنه؛ هر
    OperationalError دیگه (یا شکستِ تلاشِ آخر) دست‌نخورده بالا می‌ره.
    """
    for attempt in range(retries):
        try:
            return fn()
        except OperationalError as exc:
            if 'locked' not in str(exc).lower() or attempt == retries - 1:
                raise
            time.sleep(base_delay * (attempt + 1))

# عمداً به notifications.models.Notification.TYPE_CHOICES اضافه نشده: آن
# choices فقط رویدادهاییه که notifications به‌عنوانِ 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
        _retry_on_sqlite_lock(lambda: delivery.save(update_fields=['status']))
        return

    delivery.status = CampaignDelivery.STATUS_FAILED
    delivery.error_message = str(error)[:300]
    _retry_on_sqlite_lock(lambda: 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 می‌شه.
    """
    def _create():
        with transaction.atomic():
            return CampaignDelivery.objects.create(
                campaign=campaign, user_id=user_id, channel=campaign.channel,
            )

    try:
        return _retry_on_sqlite_lock(_create)
    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
            def _notify_and_mark_sent():
                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)

            try:
                _retry_on_sqlite_lock(_notify_and_mark_sent)
            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}")