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, PromotionTransaction
 
 class PromotionAdmin(admin.ModelAdmin):
-    list_display = ['user_uuid', 'plan', 'promotion_amount', 'recipient', 'event', 'created_at']
+    list_display = ['user_uuid', 'plan', 'promotion_amount', 'recipient', 'event', 'state', 'created_at']
+
+class PromotionTransactionAdmin(admin.ModelAdmin):
+    list_display = ['user_uuid', 'event_label', 'transaction_type', 'amount', 'state', 'promotion', 'reverses', 'created_at']
+    list_filter = ['transaction_type', 'state']
+    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(PromotionTransaction, PromotionTransactionAdmin)
 

apps/promotions/migrations/0011_promotiontransaction.py

@@ -0,0 +1,43 @@
+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.CreateModel(
+            name='PromotionTransaction',
+            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)),
+                ('transaction_type', models.IntegerField(choices=[(1, 'payout'), (2, 'rollback')], db_index=True)),
+                ('holder_wallet', models.UUIDField()),
+                ('destination_wallet', models.UUIDField()),
+                ('amount', models.IntegerField(default=0)),
+                ('state', models.IntegerField(choices=[(1, 'created'), (2, 'delayed'), (3, 'pending'), (4, 'incomplete'), (5, 'success'), (6, 'failed'), (7, 'expected_failure')], db_index=True, default=1)),
+                ('external_uuid', models.UUIDField(default=uuid.uuid4, editable=False)),
+                ('promotion', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='transactions', to='promotions.promotion')),
+                ('reverses', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='reversed_by', to='promotions.promotiontransaction')),
+            ],
+            options={
+                'ordering': ['created_at'],
+            },
+        ),
+        migrations.AddConstraint(
+            model_name='promotiontransaction',
+            constraint=models.UniqueConstraint(fields=('promotion', 'transaction_type'), name='unique_promotion_transaction_type'),
+        ),
+        migrations.AddConstraint(
+            model_name='promotiontransaction',
+            constraint=models.UniqueConstraint(condition=models.Q(('transaction_type', 2)), fields=('user_uuid', 'event_label'), name='unique_promotion_rollback'),
+        ),
+    ]

apps/promotions/models.py

@@ -364,6 +364,24 @@ class Recipient(BaseModel):
             base_amount=base_amount,
         )
         if recipient and promotion_amount and created:
+            def _pending_rollback():
+                return PromotionTransaction.objects.filter(
+                    user_uuid=recipient, event_label=event.label,
+                    transaction_type=PromotionTransaction.TypeChoices.ROLLBACK,
+                ).exclude(state=PaymentStateChoices.SUCCESS).first()
+
+            pending_rollback = _pending_rollback()
+            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)
+                PromotionTransaction.objects.filter(pk=pending_rollback.pk).update(
+                    promotion=promotion, amount=0, state=PaymentStateChoices.SUCCESS,
+                    updated_at=timezone.now())
+                promotion.refresh_from_db()
+                return promotion
+
             try:
                 with transaction.atomic():
                     reserved = self.plan.reserve_promotion_amount(promotion_amount)
@@ -376,6 +394,11 @@ 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). Reverse it now that it has settled.
+            if _pending_rollback() is not None:
+                rollback_promotion_payout(recipient, event.label)
+
             return promotion
         return None
 
@@ -497,60 +520,38 @@ class Promotion(BaseModel):
             error_message='Cannot promote. not in current state'
         )
 
-        payment_uuid = str(self.uuid)
-
-        payee_wallet = self.recipient.get_wallet_category_uuid()
-
-        data = {
-            "uuid": payment_uuid,
-            "payee": str(self.user_uuid),
-            "payee_type": 1,
-            "payee_wallet": payee_wallet,
-            "amount": self.promotion_amount,
-            "details": {
-                'description': str(_(self.recipient.label)),
-                'reference_id': str(self.pk),
-                'application_details_url': ''
-            },
-        }
+        # The wallet transfer itself is recorded and driven by a PromotionTransaction
+        # ledger row (mirrors advertising's AdPayment / escrow's EscrowWalletPayment),
+        # so every promotion payout has an auditable record and can be reversed.
+        payout, _created = PromotionTransaction.objects.get_or_create(
+            promotion=self,
+            transaction_type=PromotionTransaction.TypeChoices.PAYOUT,
+            defaults=dict(
+                user_uuid=self.user_uuid,
+                event_label=self.event.label if self.event_id else '',
+                holder_wallet=settings.WALLET_PROMOTIONS_CREDIT,
+                destination_wallet=self.recipient.get_wallet_category_uuid(),
+                amount=self.promotion_amount,
+            ),
+        )
 
         try:
-            submit_response = deposit_to_user_wallet_submit(settings.WALLET_PROMOTIONS_CREDIT, data)
-            if not submit_response.uuid:
-                raise Exception('Failed to submit promotion. 1')
-
-        except Exception as e:
-            self.change_state(
-                from_states=[self.state],
-                to_state=PaymentStateChoices.FAILED,
-                same_ok=False,
-                raise_exception=True,
-                error_message='Failed to update state promotions'
-            )
-            # return False
-            raise Exception('Failed to submit promotion. 2')
-
-        try:
-            verify_response = deposit_to_user_wallet_verify(settings.WALLET_PROMOTIONS_CREDIT, submit_response.uuid)
-            if not verify_response.uuid:
-                raise Exception('Failed to verify promotion. 1')
-
-        except Exception as e:
+            transferred = payout.execute()
+        except Exception:
             self.change_state(
                 from_states=[self.state],
                 to_state=PaymentStateChoices.EXPECTED_FAILURE,
-                same_ok=False,
+                same_ok=True,
                 raise_exception=True,
-                error_message='Failed to update state'
+                error_message='Failed to update state',
             )
-            # return False
-            raise Exception('Failed to verify promition. 2')
+            raise Exception('Failed to pay promotion')
 
-        if verify_response.state != 5:
-            to_pay_failed = self.change_state(
+        if not transferred:
+            self.change_state(
                 from_states=[self.state],
                 to_state=PaymentStateChoices.FAILED,
-                same_ok=False,
+                same_ok=True,
                 raise_exception=True,
                 error_message='Failed to update state payment',
             )
@@ -584,3 +585,192 @@ 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 is_rolled_back(self):
+        return self.transactions.filter(
+            transaction_type=PromotionTransaction.TypeChoices.ROLLBACK,
+            state=PaymentStateChoices.SUCCESS,
+        ).exists()
+
+
+IN_FLIGHT_PAYMENT_STATES = (
+    PaymentStateChoices.CREATED,
+    PaymentStateChoices.PENDING,
+    PaymentStateChoices.DELAYED,
+    PaymentStateChoices.INCOMPLETE,
+)
+
+
+class PromotionTransaction(BaseModel):
+    """Ledger of every wallet transfer the promotions service makes -- the
+    payout for a promotion, and any later reversal of it. Mirrors advertising's
+    ``AdPayment`` and escrow's ``EscrowWalletPayment``: each row drives one
+    external wallet call through its own CREATED -> PENDING -> SUCCESS/FAILED
+    state, so the money movement is auditable and reversible.
+
+    ``user_uuid`` / ``event_label`` are denormalised onto the row so a ROLLBACK
+    can be recorded before the async payout task has even created its
+    ``Promotion`` (see :func:`rollback_promotion_payout` and
+    ``Recipient.promote``).
+    """
+
+    class TypeChoices(models.IntegerChoices):
+        PAYOUT = 1, _('payout')      # promotions credit  -> recipient wallet (transit or user)
+        ROLLBACK = 2, _('rollback')  # advertising transit -> promotions credit
+
+    promotion = models.ForeignKey(Promotion, on_delete=models.SET_NULL, null=True, blank=True,
+                                  related_name='transactions')
+    user_uuid = models.UUIDField(db_index=True)
+    event_label = models.CharField(max_length=255, db_index=True)
+    transaction_type = models.IntegerField(choices=TypeChoices.choices, db_index=True)
+    holder_wallet = models.UUIDField()        # 1st arg to deposit_to_user_wallet_submit
+    destination_wallet = models.UUIDField()   # data['payee_wallet']
+    amount = models.IntegerField(default=0)
+    state = models.IntegerField(choices=PaymentStateChoices.choices,
+                                default=PaymentStateChoices.CREATED, db_index=True)
+    external_uuid = models.UUIDField(default=uuid.uuid4, editable=False)  # idempotency key to the wallet service
+    reverses = models.ForeignKey('self', on_delete=models.PROTECT, null=True, blank=True,
+                                 related_name='reversed_by')
+
+    class Meta:
+        ordering = ['created_at']
+        constraints = [
+            models.UniqueConstraint(fields=['promotion', 'transaction_type'],
+                                    name='unique_promotion_transaction_type'),
+            models.UniqueConstraint(fields=['user_uuid', 'event_label'],
+                                    condition=models.Q(transaction_type=2),
+                                    name='unique_promotion_rollback'),
+        ]
+
+    def __str__(self):
+        return f"{self.get_transaction_type_display()} {self.amount} ({self.event_label})"
+
+    def _set_state(self, state):
+        PromotionTransaction.objects.filter(pk=self.pk).update(state=state, updated_at=timezone.now())
+        self.state = state
+
+    def execute(self):
+        """Run the wallet transfer once. Returns True only on a confirmed
+        transfer; leaves the row FAILED / EXPECTED_FAILURE (and re-raises) on
+        error, same as the promote()/AdPayment style. No automatic retry."""
+        if self.state == PaymentStateChoices.SUCCESS:
+            return True
+        if self.state != PaymentStateChoices.CREATED:
+            return False
+        if not self.amount:
+            self._set_state(PaymentStateChoices.SUCCESS)
+            return True
+
+        self._set_state(PaymentStateChoices.PENDING)
+
+        data = {
+            "uuid": str(self.external_uuid),
+            "payee": str(self.user_uuid),
+            "payee_type": 1,
+            "payee_wallet": str(self.destination_wallet),
+            "amount": self.amount,
+            "details": {
+                'description': str(self.get_transaction_type_display()),
+                'reference_id': str(self.pk),
+                'application_details_url': '',
+            },
+        }
+
+        try:
+            submit_response = deposit_to_user_wallet_submit(str(self.holder_wallet), data)
+            if not getattr(submit_response, 'uuid', None):
+                raise Exception('wallet submit returned no uuid')
+        except Exception:
+            self._set_state(PaymentStateChoices.FAILED)
+            raise
+
+        try:
+            verify_response = deposit_to_user_wallet_verify(str(self.holder_wallet), submit_response.uuid)
+            if not getattr(verify_response, 'uuid', None):
+                raise Exception('wallet verify returned no uuid')
+        except Exception:
+            self._set_state(PaymentStateChoices.EXPECTED_FAILURE)
+            raise
+
+        if verify_response.state != 5:
+            self._set_state(PaymentStateChoices.FAILED)
+            return False
+
+        self._set_state(PaymentStateChoices.SUCCESS)
+        return True
+
+
+def rollback_promotion_payout(user_uuid, event_label):
+    """Reverse a promotion payout for ``(user_uuid, event_label)`` back out of
+    the advertising transit wallet into the promotions credit wallet.
+
+    The promotion stays consumed -- the user cannot earn it again -- only the
+    money moves, and only if the payout actually landed in the transit wallet.
+    Idempotent. Returns ``(status, amount)`` where status is one of
+    ``'reversed'`` (money moved), ``'deferred'`` (payout not processed yet --
+    ``Recipient.promote()`` will suppress it), ``'nothing'`` (nothing to
+    reverse: zero payout, or paid straight to the user's wallet).
+    """
+    with transaction.atomic():
+        existing = PromotionTransaction.objects.select_for_update().filter(
+            user_uuid=user_uuid, event_label=event_label,
+            transaction_type=PromotionTransaction.TypeChoices.ROLLBACK,
+        ).first()
+        if existing is not None and existing.state == PaymentStateChoices.SUCCESS:
+            return ('reversed' if existing.amount else 'nothing', existing.amount)
+
+        payout = PromotionTransaction.objects.select_for_update().filter(
+            user_uuid=user_uuid, event_label=event_label,
+            transaction_type=PromotionTransaction.TypeChoices.PAYOUT,
+        ).first()
+
+        payout_in_flight = payout is not None and payout.state in IN_FLIGHT_PAYMENT_STATES
+        promotion_settled_without_payout = payout is None and Promotion.objects.filter(
+            user_uuid=user_uuid, event__label=event_label, state=PaymentStateChoices.SUCCESS,
+        ).exists()
+
+        if payout is not None and payout.state == PaymentStateChoices.SUCCESS \
+                and payout.promotion_id and payout.promotion.is_rollbackable() and payout.amount:
+            amount = payout.amount
+        else:
+            amount = 0
+
+        promo = payout.promotion if payout is not None else None
+
+        rollback, created = PromotionTransaction.objects.get_or_create(
+            user_uuid=user_uuid, event_label=event_label,
+            transaction_type=PromotionTransaction.TypeChoices.ROLLBACK,
+            defaults=dict(
+                promotion=promo,
+                holder_wallet=settings.WALLET_ADVERTISING_TRANSIT,
+                destination_wallet=settings.WALLET_PROMOTIONS_CREDIT,
+                amount=amount,
+                reverses=payout,
+            ),
+        )
+        if not created and rollback.state != PaymentStateChoices.SUCCESS:
+            # refresh the amount/links now that the payout may have appeared
+            PromotionTransaction.objects.filter(pk=rollback.pk).exclude(
+                state=PaymentStateChoices.SUCCESS).update(
+                promotion=promo, amount=amount, reverses=payout, updated_at=timezone.now())
+            rollback.amount = amount
+
+        if payout is None and not promotion_settled_without_payout:
+            return ('deferred', 0)          # honoured later by Recipient.promote()
+        if payout_in_flight:
+            return ('deferred', 0)
+        if amount == 0:
+            rollback._set_state(PaymentStateChoices.SUCCESS)
+            return ('nothing', 0)
+
+        rollback.execute()                  # raises on wallet failure -> caller aborts + retries
+        return ('reversed', 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, EventSaver, get_event_status_for_user, rollback_promotion_payout
 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,30 @@ 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()
+        state, amount = rollback_promotion_payout(user.uuid, event_label)
+
+        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()