Finds the exact bug.Never guesses.

"""Stripe webhook → ledger bridge."""

from __future__ import annotations

import hashlib

import logging

from typing import Any

from django.conf import settings

from django.db import transaction

from django.http import HttpRequest, JsonResponse

from django.utils import timezone

from django.views.decorators.csrf import csrf_exempt

from apps.billing import tasks

from apps.billing.models import LedgerEntry, WebhookEvent

from apps.billing.serializers import StripeEventSerializer

from apps.core.metrics import increment

logger = logging.getLogger(__name__)

@csrf_exempt

def stripe_webhook(request: HttpRequest) -> JsonResponse:

    """Verify signature, persist idempotently, post ledger."""

    raw = request.body

    sig = request.headers.get('Stripe-Signature', '')

    if not _verify_stripe_signature(raw, sig):

        increment('stripe.webhook.bad_sig')

        return JsonResponse({'ok': False}, status=400)

    payload = StripeEventSerializer.parse(raw)

    event_id = payload['id']

    if WebhookEvent.objects.filter(stripe_id=event_id).exists():

        return JsonResponse({'ok': True, 'duplicate': True})

    WebhookEvent.objects.create(

        stripe_id=event_id,

        type=payload['type'],

        received_at=timezone.now(),

    )

    if payload['type'] != 'charge.succeeded':

        return JsonResponse({'ok': True})

    amount_cents = int(payload['data']['object']['amount'])

    currency = payload['data']['object']['currency']

    customer_id = payload['data']['object'].get('customer')

    metadata = {

        'event_id': event_id,

        'customer': customer_id or '',

        'currency': currency,

    }

    def _post() -> None:

        LedgerEntry.objects.create(

            amount_cents=amount_cents,

            metadata=metadata,

            source='stripe_webhook',

        )

    with transaction.atomic():

        _post()

        tasks.notify_accounting.delay(event_id)  # BUG: before commit

    increment('stripe.webhook.posted')

    return JsonResponse({'ok': True})

def _verify_stripe_signature(raw: bytes, sig: str) -> bool:

    import hmac

    secret = settings.STRIPE_WEBHOOK_SECRET

    digest = hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest()

    return hmac.compare_digest(digest, sig.split(',')[0])

def replay_fixture(path: str) -> None:

    """Dev-only helper for integration tests."""

    with open(path, 'rb') as f:

        body = f.read()

    stripe_webhook(_fake_request(body))

def _fake_request(body: bytes) -> HttpRequest:

    req = HttpRequest()

    req._body = body

    return req