Payment and Webhook Workflows

class Payment(models.Model):
    class Status(models.TextChoices):
        CREATED = "created", "Created"
        PROCESSING = "processing", "Processing"
        REQUIRES_ACTION = "requires_action", "Requires action"
        SUCCEEDED = "succeeded", "Succeeded"
        FAILED = "failed", "Failed"
        REFUNDED = "refunded", "Refunded"
        DISPUTED = "disputed", "Disputed"

    # Only the provider may cause these. The transition method enforces it.
    PROVIDER_ONLY = {Status.SUCCEEDED, Status.FAILED, Status.DISPUTED}

    order = models.ForeignKey("orders.Order", on_delete=models.PROTECT,
                              related_name="payments")
    status = models.CharField(max_length=16, choices=Status.choices,
                              default=Status.CREATED, db_index=True)

    # Decimal, not float: 0.10 has no exact binary representation, and the
    # error compounds across totals into reconciliation failures.
    amount = models.DecimalField(max_digits=12, decimal_places=2)
    currency = models.CharField(max_length=3)          # ISO 4217

    # The join key for every webhook, refund and dispute that follows.
    provider = models.CharField(max_length=32)
    provider_reference = models.CharField(max_length=128, blank=True, db_index=True)

    class Meta:
        constraints = [
            models.CheckConstraint(
                condition=models.Q(amount__gt=0), name="payment_amount_positive"
            ),
            # One provider reference maps to one payment, per provider.
            models.UniqueConstraint(
                fields=["provider", "provider_reference"],
                condition=~models.Q(provider_reference=""),
                name="uniq_provider_reference",
            ),
        ]

    def transition_to(self, target, *, source, event=None):
        """`source` is "provider" or "internal". The distinction is the
        section's core principle, expressed as code rather than as a
        convention somebody has to remember."""
        if target in self.PROVIDER_ONLY and source != "provider":
            raise PermissionDenied(
                f"{target} requires a verified provider event, not {source}"
            )
        ...                                            # then the usual checks


class PaymentAttempt(models.Model):
    """One payment, many attempts: a card declined twice then accepted is
    one intent with three attempts, not three payments."""
    payment = models.ForeignKey(Payment, on_delete=models.CASCADE,
                                related_name="attempts")
    provider_reference = models.CharField(max_length=128, db_index=True)
    outcome = models.CharField(max_length=32)
    failure_code = models.CharField(max_length=64, blank=True)
    occurred_at = models.DateTimeField(default=timezone.now)

import hashlib, hmac, json, time
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

TOLERANCE_SECONDS = 300


@csrf_exempt          # a provider has no CSRF token; the signature replaces it
@require_POST
def provider_webhook(request):
    # 1. RAW BYTES FIRST. Django's docs: accessing body after reading the
    #    stream produces RawPostDataException — and any parse changes them.
    raw = request.body

    signature = request.headers.get("X-Signature", "")
    timestamp = request.headers.get("X-Timestamp", "")

    # 2. Freshness. A signature stays valid forever; this is what stops a
    #    captured event being replayed next month.
    try:
        if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
            return HttpResponseForbidden("stale timestamp")
    except ValueError:
        return HttpResponseForbidden("bad timestamp")

    # 3. Constant-time comparison, against every currently valid secret so
    #    that rotation does not need an outage.
    signed_payload = timestamp.encode() + b"." + raw
    if not any(
        hmac.compare_digest(
            hmac.new(secret, signed_payload, hashlib.sha256).hexdigest(), signature
        )
        for secret in settings.WEBHOOK_SECRETS          # [current, previous]
    ):
        return HttpResponseForbidden("bad signature")

    # 4. Only now is this data trustworthy enough to parse.
    event = json.loads(raw)

    # 5. Persist and acknowledge FAST. Processing happens in a worker, so a
    #    slow handler cannot cause provider timeouts and a retry storm.
    row, created = WebhookEvent.objects.get_or_create(
        provider="acme", event_id=event["id"],
        defaults={"payload": event, "received_at": timezone.now()},
    )
    if created:
        transaction.on_commit(lambda: process_webhook_event.delay(row.id))

    return HttpResponse(status=200)

STATUS_RANK = {
    "created": 0, "processing": 1, "requires_action": 1,
    "succeeded": 2, "failed": 2, "refunded": 3, "disputed": 4,
}


@shared_task(bind=True, max_retries=5, retry_backoff=True, retry_jitter=True)
def process_webhook_event(self, event_row_id):
    row = WebhookEvent.objects.get(pk=event_row_id)

    if row.processed_at is not None:
        return                       # already applied — a duplicate task run

    payload = row.payload
    payment = Payment.objects.filter(
        provider=row.provider, provider_reference=payload["payment_reference"]
    ).first()

    if payment is None:
        # The event may have overtaken our own record of the attempt. Retry
        # with backoff rather than dropping it — but bound the attempts.
        raise self.retry(countdown=30)

    with transaction.atomic():
        # Lock, then re-read: two events for the same payment can be in
        # flight at once, and this is the same race as any state machine.
        payment = Payment.objects.select_for_update().get(pk=payment.pk)

        incoming = payload["status"]
        if STATUS_RANK[incoming] <= STATUS_RANK[payment.status]:
            # Stale or already applied. Mark the event handled and stop —
            # it is not an error, and it must not be retried.
            row.processed_at = timezone.now()
            row.outcome = "ignored_stale"
            row.save(update_fields=["processed_at", "outcome"])
            return

        payment.transition_to(incoming, source="provider", event=row)

        PaymentEvent.objects.create(payment=payment, event=row, to_state=incoming)
        row.processed_at = timezone.now()
        row.outcome = "applied"
        row.save(update_fields=["processed_at", "outcome"])

        # Irreversible work, only after the commit.
        if incoming == "succeeded":
            transaction.on_commit(lambda: fulfil_order.delay(payment.order_id))
            transaction.on_commit(lambda: send_receipt.delay(payment.id))

class Command(BaseCommand):
    help = "Find payments the provider has resolved but we have not heard about."

    def add_arguments(self, parser):
        parser.add_argument("--stale-minutes", type=int, default=30)
        parser.add_argument("--apply", action="store_true")

    def handle(self, *args, **options):
        cutoff = timezone.now() - timedelta(minutes=options["stale_minutes"])

        # A payment stuck non-terminal past its window is either genuinely
        # in progress or missing a webhook. Only the provider can say which.
        stale = Payment.objects.filter(
            status__in=[Payment.Status.PROCESSING, Payment.Status.REQUIRES_ACTION],
            updated_at__lt=cutoff,
        ).exclude(provider_reference="")

        drift = 0
        for payment in stale.iterator(chunk_size=200):
            remote = provider_client.fetch(payment.provider, payment.provider_reference)

            if remote.status == payment.status:
                continue                       # genuinely still in progress

            drift += 1
            self.stdout.write(
                f"{payment.provider_reference}: ours={payment.status} "
                f"theirs={remote.status}"
            )

            if options["apply"]:
                # The SAME transition path a webhook uses. A second writer of
                # payment state is a second place for the rules to diverge.
                payment.transition_to(remote.status, source="provider",
                                      event=remote.as_event())

        # A rising count is a webhook-delivery alarm, not just a repair job.
        self.stderr.write(
            self.style.WARNING(f"{drift} payments out of sync")
            if drift else self.style.SUCCESS("in sync")
        )
        if drift:
            raise CommandError(f"{drift} payments reconciled", returncode=3)