FEATURE(promotions): reversible promotion payouts to the advertising transit wallet

When the advertising service discards what a promotion paid for (e.g. a
captured billboard deleted while still pending approval), the promotion
money sitting in the advertising transit wallet needs to go back to the
promotions credit wallet. The promotion itself stays consumed -- only the
money is returned -- and only payouts that landed in the transit wallet
are reversible (a payout straight to the user's wallet is the user's).

- Promotion.rolled_back_at + rollback_to_credit(): row-locked, CAS-stamped,
  idempotent transfer transit -> credit for SUCCESS/transit-destined payouts.
- PromotionRollback model: keyed (user, event_label) -- all the advertising
  side knows. Handles the async race (payout runs in a Celery task, so the
  Promotion may not exist yet): Recipient.promote() checks for an unsettled
  request before paying (suppresses the payout) and after (reverses a
  payout that landed mid-request).
- POST .../event/<event_label>/rollback/ -> {status: reversed|deferred|nothing, amount}.
- migration 0011 (hand-written; verified via makemigrations --dry-run + check).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

apps/promotions/admin.py

@@ -2,10 +2,15 @@ from functools import update_wrapper
 
 from django.contrib import admin
 
-from .models import Promotion, Plan, Event, Recipient, EventSaver, AllowedUser
+from .models import Promotion, Plan, Event, Recipient, EventSaver, AllowedUser, PromotionRollback
 
 class PromotionAdmin(admin.ModelAdmin):
-    list_display = ['user_uuid', 'plan', 'promotion_amount', 'recipient', 'event', 'created_at']
+    list_display = ['user_uuid', 'plan', 'promotion_amount', 'recipient', 'event', 'rolled_back_at', 'created_at']
+
+class PromotionRollbackAdmin(admin.ModelAdmin):
+    list_display = ['user_uuid', 'event_label', 'amount', 'settled_at', 'promotion', 'created_at']
+    list_filter = ['settled_at']
+    search_fields = ['user_uuid', 'event_label']
 
 class PlanAdmin(admin.ModelAdmin):
     list_display = ['uuid', 'title', 'balance']
@@ -19,4 +24,5 @@ admin.site.register(Event)
 admin.site.register(EventSaver)
 admin.site.register(Recipient)
 admin.site.register(AllowedUser, AllowedUserAdmin)
+admin.site.register(PromotionRollback, PromotionRollbackAdmin)
 

apps/promotions/migrations/0011_promotion_rolled_back_at_promotionrollback.py

@@ -0,0 +1,39 @@
+import uuid
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('promotions', '0010_recipient_wallet_destination'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='promotion',
+            name='rolled_back_at',
+            field=models.DateTimeField(blank=True, db_index=True, null=True),
+        ),
+        migrations.CreateModel(
+            name='PromotionRollback',
+            fields=[
+                ('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, primary_key=True, serialize=False, unique=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
+                ('updated_at', models.DateTimeField(auto_now=True, db_index=True)),
+                ('user_uuid', models.UUIDField(db_index=True)),
+                ('event_label', models.CharField(db_index=True, max_length=255)),
+                ('amount', models.IntegerField(default=0)),
+                ('settled_at', models.DateTimeField(blank=True, db_index=True, null=True)),
+                ('promotion', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='rollback_requests', to='promotions.promotion')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.AddConstraint(
+            model_name='promotionrollback',
+            constraint=models.UniqueConstraint(fields=('user_uuid', 'event_label'), name='unique_promotion_rollback'),
+        ),
+    ]

apps/promotions/models.py

@@ -364,6 +364,21 @@ class Recipient(BaseModel):
             base_amount=base_amount,
         )
         if recipient and promotion_amount and created:
+            pending_rollback = PromotionRollback.objects.filter(
+                user_uuid=recipient, event_label=event.label, settled_at__isnull=True,
+            ).first()
+            if pending_rollback is not None:
+                # The advertising service already asked to reverse this promotion
+                # before it was even paid -- don't move the money at all.
+                promotion.change_state(from_states=[promotion.state],
+                                       to_state=PaymentStateChoices.SUCCESS, same_ok=True)
+                Promotion.objects.filter(pk=promotion.pk, rolled_back_at__isnull=True).update(
+                    rolled_back_at=timezone.now(), updated_at=timezone.now())
+                PromotionRollback.objects.filter(pk=pending_rollback.pk, settled_at__isnull=True).update(
+                    promotion=promotion, amount=0, settled_at=timezone.now(), updated_at=timezone.now())
+                promotion.refresh_from_db()
+                return promotion
+
             try:
                 with transaction.atomic():
                     reserved = self.plan.reserve_promotion_amount(promotion_amount)
@@ -376,6 +391,15 @@ class Recipient(BaseModel):
                                        same_ok=True)
                 promotion.refresh_from_db()
 
+            # A rollback request may have arrived while the payout above was in
+            # flight (so the pre-check missed it). Now that it has settled,
+            # reverse the money.
+            late_rollback = PromotionRollback.objects.filter(
+                user_uuid=recipient, event_label=event.label, settled_at__isnull=True,
+            ).first()
+            if late_rollback is not None:
+                late_rollback.resolve()
+
             return promotion
         return None
 
@@ -436,6 +460,10 @@ class Promotion(BaseModel):
     base_amount = models.IntegerField(null=True, blank=True)
     promotion_amount = models.IntegerField(null=True, blank=True)
     data = models.JSONField(null=True, blank=True)
+    # Set once the promotion's money has been pulled back out of the advertising
+    # transit wallet (see rollback_to_credit / PromotionRollback). The promotion
+    # stays SUCCESS and still counts as "used" -- only the money is returned.
+    rolled_back_at = models.DateTimeField(null=True, blank=True, db_index=True)
 
     objects = PromotionQuerySet.as_manager()
 
@@ -584,3 +612,136 @@ class Promotion(BaseModel):
                     error_message='Failed to update state payment',
                 )
                 return False
+
+    def is_rollbackable(self):
+        """A promotion payout can only be reversed if it actually landed in the
+        advertising transit wallet. A payout made straight to the user's income
+        wallet is the user's money and is never clawed back."""
+        return bool(
+            self.recipient_id
+            and self.recipient.wallet_destination == WalletDestinationChoices.ADVERTISING_TRANSIT
+        )
+
+    def rollback_to_credit(self):
+        """Move an already-paid promotion's money from the advertising transit
+        wallet back to the promotions credit wallet.
+
+        Called (via the rollback endpoint) when the advertising service discards
+        whatever the promotion paid for -- e.g. a captured billboard deleted
+        while still pending approval. The Promotion row stays SUCCESS and keeps
+        counting as "used"; the user does not get to earn it again. Only the
+        money moves.
+
+        Returns the amount returned (0 when there is nothing to reverse:
+        not yet paid, already rolled back, zero amount, or a payout that went
+        straight to the user's wallet). Idempotent -- safe to call repeatedly
+        and concurrently.
+        """
+        with transaction.atomic():
+            promo = Promotion.objects.select_for_update().get(pk=self.pk)
+
+            if promo.rolled_back_at is not None:
+                return 0
+            if promo.state != PaymentStateChoices.SUCCESS:
+                return 0
+            if not promo.promotion_amount:
+                # balance_holder plans / zero payouts never moved any money
+                Promotion.objects.filter(pk=promo.pk, rolled_back_at__isnull=True).update(
+                    rolled_back_at=timezone.now(), updated_at=timezone.now())
+                self.rolled_back_at = timezone.now()
+                return 0
+            if not promo.is_rollbackable():
+                return 0
+
+            amount = promo.promotion_amount
+            payment_uuid = str(uuid.uuid4())
+            data = {
+                "uuid": payment_uuid,
+                "payee": str(promo.user_uuid),
+                "payee_type": 1,
+                "payee_wallet": settings.WALLET_PROMOTIONS_CREDIT,
+                "amount": amount,
+                "details": {
+                    'description': str(_('promotion rollback')) + f": {promo.recipient.label}",
+                    'reference_id': str(promo.pk),
+                    'application_details_url': '',
+                },
+            }
+
+            submit_response = deposit_to_user_wallet_submit(settings.WALLET_ADVERTISING_TRANSIT, data)
+            if not getattr(submit_response, 'uuid', None):
+                raise Exception('Failed to submit promotion rollback')
+
+            verify_response = deposit_to_user_wallet_verify(
+                settings.WALLET_ADVERTISING_TRANSIT, submit_response.uuid)
+            if not getattr(verify_response, 'uuid', None):
+                raise Exception('Failed to verify promotion rollback')
+            if verify_response.state != 5:
+                raise Exception('Promotion rollback not confirmed by wallet service')
+
+            claimed = Promotion.objects.filter(pk=promo.pk, rolled_back_at__isnull=True).update(
+                rolled_back_at=timezone.now(), updated_at=timezone.now())
+            if not claimed:
+                # someone else stamped it between our checks -- but we already
+                # moved the money. This should be impossible under select_for_update;
+                # surface it loudly rather than silently double-pay on a retry.
+                raise Exception('Promotion rollback raced after wallet transfer')
+
+            self.rolled_back_at = timezone.now()
+            return amount
+
+
+class PromotionRollback(BaseModel):
+    """A request from the advertising service to pull a promotion's money back
+    out of the advertising transit wallet (the thing it paid for was removed
+    before it counted -- e.g. a captured billboard deleted while pending
+    approval).
+
+    Keyed by (user, event label) because that is all the advertising side
+    knows. The matching Promotion may not exist yet when the request arrives
+    (the payout runs in an async task), so this row is also checked by
+    Recipient.promote() before it pays out: an unsettled request there means
+    the money is never sent in the first place.
+    """
+    user_uuid = models.UUIDField(db_index=True)
+    event_label = models.CharField(max_length=255, db_index=True)
+    promotion = models.ForeignKey(
+        Promotion, on_delete=models.SET_NULL, null=True, blank=True, related_name='rollback_requests')
+    amount = models.IntegerField(default=0)
+    # Set once we've either moved the money back or confirmed there was nothing
+    # to move (payout suppressed before it happened, or not rollbackable).
+    settled_at = models.DateTimeField(null=True, blank=True, db_index=True)
+
+    class Meta:
+        constraints = [
+            models.UniqueConstraint(fields=['user_uuid', 'event_label'], name='unique_promotion_rollback'),
+        ]
+
+    def __str__(self):
+        return f"rollback {self.event_label} / {self.user_uuid}"
+
+    def resolve(self):
+        """Try to settle this request against a promotion that already exists.
+        If none exists yet, stay unsettled -- Recipient.promote() will pick it
+        up when (if) the payout is processed."""
+        if self.settled_at is not None:
+            return self.amount
+
+        promotion = Promotion.objects.filter(
+            user_uuid=self.user_uuid, event__label=self.event_label,
+        ).order_by('-created_at').first()
+
+        if promotion is None:
+            return 0
+
+        if promotion.state in (PaymentStateChoices.CREATED, PaymentStateChoices.PENDING,
+                               PaymentStateChoices.DELAYED, PaymentStateChoices.INCOMPLETE):
+            # payout is in flight in the worker; let Recipient.promote() settle it
+            return 0
+
+        amount = promotion.rollback_to_credit()
+        PromotionRollback.objects.filter(pk=self.pk, settled_at__isnull=True).update(
+            promotion=promotion, amount=amount, settled_at=timezone.now(), updated_at=timezone.now())
+        self.amount = amount
+        self.settled_at = timezone.now()
+        return amount

apps/promotions/serializers.py

@@ -90,6 +90,15 @@ class PromotionStatusSerializer(serializers.Serializer):
     promotion_amount = serializers.IntegerField(read_only=True, allow_null=True)
 
 
+class PromotionRollbackSerializer(serializers.Serializer):
+    event_label = serializers.CharField(read_only=True)
+    # reversed  -> money moved back out of the advertising transit wallet
+    # deferred  -> payout not settled yet; it will be suppressed when it runs
+    # nothing   -> nothing to reverse (already rolled back, or paid to the user)
+    status = serializers.ChoiceField(read_only=True, choices=['reversed', 'deferred', 'nothing'])
+    amount = serializers.IntegerField(read_only=True)
+
+
 class UserPlanSerializer(serializers.ModelSerializer):
     recipients = UserRecipientSerializer(many=True, read_only=True)
     class Meta:

apps/promotions/views_application.py

@@ -12,7 +12,7 @@ from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRe
 from apps.gooyal_oauth2.utils import get_application
 from utils.clients.accounts_client import get_user_info
 from utils.exceptions import UnprocessableEntity
-from .models import Plan, Promotion, EventSaver, get_event_status_for_user
+from .models import Plan, Promotion, PromotionRollback, EventSaver, get_event_status_for_user
 from .serializers import (
     PlanSerializer,
     PromotionSerializer,
@@ -20,6 +20,7 @@ from .serializers import (
     PromoteSerializer,
     UserPlanSerializer,
     PromotionStatusSerializer,
+    PromotionRollbackSerializer,
 )
 from .tasks import analyze_event_task
 from ..users.models import User
@@ -144,6 +145,40 @@ class ApplicationEventViewSet(
         serializer = self.get_serializer(get_event_status_for_user(user.uuid, event_label))
         return Response(serializer.data)
 
+    @action(
+        detail=False,
+        methods=['POST'],
+        url_path=r'(?P<event_label>.+)/rollback',
+        serializer_class=PromotionRollbackSerializer,
+    )
+    def rollback(self, request, user_uuid=None, event_label=None):
+        """Pull a promotion's payout back out of the advertising transit wallet.
+
+        Idempotent. The promotion stays "used" -- only the money is returned,
+        and only if it landed in the transit wallet (a payout straight to the
+        user's wallet is not reversible). If the payout has not been processed
+        yet the request is recorded and the payout is suppressed when it runs.
+        """
+        user = self._resolve_user()
+        rb, _created = PromotionRollback.objects.get_or_create(
+            user_uuid=user.uuid, event_label=event_label,
+        )
+        amount = rb.resolve()
+
+        if rb.settled_at is None:
+            state = 'deferred'
+        elif amount:
+            state = 'reversed'
+        else:
+            state = 'nothing'
+
+        serializer = self.get_serializer({
+            'event_label': event_label,
+            'status': state,
+            'amount': amount,
+        })
+        return Response(serializer.data)
+
     def perform_create(self, serializer: EventSerializer):
         user = self._resolve_user()