(SP: 1) [Stripe Webhook] Add multi-instance claim lock (claimed_at/by…

frontend/app/api/shop/webhooks/stripe/route.ts

@@ -1,6 +1,7 @@
 import Stripe from 'stripe';
+import crypto from 'node:crypto';
 import { NextRequest, NextResponse } from 'next/server';
-import { and, eq, ne, or } from 'drizzle-orm';
+import { and, eq, ne, or, isNull, lt } from 'drizzle-orm';
 import { db } from '@/db';
 import { verifyWebhookSignature, retrieveCharge } from '@/lib/psp/stripe';
 import { orders, stripeEvents } from '@/db/schema';
@@ -15,6 +16,71 @@ import { markStripeAttemptFinal } from '@/lib/services/orders/payment-attempts';
 
 const REFUND_FULLNESS_UNDETERMINED = 'REFUND_FULLNESS_UNDETERMINED' as const;
 
+// P0.8: multi-instance claim/lock (no transactions; safe under parallel deliveries)
+const STRIPE_WEBHOOK_INSTANCE_ID =
+  (
+    process.env.STRIPE_WEBHOOK_INSTANCE_ID ??
+    process.env.WEBHOOK_INSTANCE_ID ??
+    ''
+  ).trim() || crypto.randomUUID().slice(0, 12);
+
+const STRIPE_EVENT_CLAIM_TTL_MS = 10 * 60 * 1000; // 10 minutes
+const STRIPE_EVENT_RETRY_AFTER_SECONDS = 10;
+
+function busyRetry() {
+  return NextResponse.json(
+    {
+      code: 'WEBHOOK_CLAIMED',
+      retryAfterSeconds: STRIPE_EVENT_RETRY_AFTER_SECONDS,
+    },
+    {
+      status: 503,
+      headers: { 'Retry-After': String(STRIPE_EVENT_RETRY_AFTER_SECONDS) },
+    }
+  );
+}
+async function tryClaimStripeEvent(
+  eventId: string
+): Promise<'claimed' | 'already_processed' | 'busy'> {
+  const now = new Date();
+  const expiresAt = new Date(now.getTime() + STRIPE_EVENT_CLAIM_TTL_MS);
+
+  const claimed = await db
+    .update(stripeEvents)
+    .set({
+      claimedAt: now,
+      claimExpiresAt: expiresAt,
+      claimedBy: STRIPE_WEBHOOK_INSTANCE_ID,
+    })
+    .where(
+      and(
+        eq(stripeEvents.eventId, eventId),
+        isNull(stripeEvents.processedAt),
+        or(
+          isNull(stripeEvents.claimedAt),
+          isNull(stripeEvents.claimExpiresAt),
+          lt(stripeEvents.claimExpiresAt, now)
+        )
+      )
+    )
+    .returning({ eventId: stripeEvents.eventId });
+
+  if (claimed.length > 0) return 'claimed';
+
+  const [row] = await db
+    .select({
+      processedAt: stripeEvents.processedAt,
+      claimExpiresAt: stripeEvents.claimExpiresAt,
+      claimedBy: stripeEvents.claimedBy,
+    })
+    .from(stripeEvents)
+    .where(eq(stripeEvents.eventId, eventId))
+    .limit(1);
+
+  if (row?.processedAt) return 'already_processed';
+  return 'busy';
+}
+
 function throwRefundFullnessUndetermined(): never {
   throw new Error(REFUND_FULLNESS_UNDETERMINED);
 }
@@ -349,10 +415,25 @@ export async function POST(request: NextRequest) {
 
   try {
     const ack = async () => {
-      await db
+      const updated = await db
         .update(stripeEvents)
         .set({ processedAt: new Date() })
-        .where(eq(stripeEvents.eventId, event.id));
+        .where(
+          and(
+            eq(stripeEvents.eventId, event.id),
+            eq(stripeEvents.claimedBy, STRIPE_WEBHOOK_INSTANCE_ID)
+          )
+        )
+        .returning({ eventId: stripeEvents.eventId });
+
+      if (updated.length === 0) {
+        logWarn('stripe_webhook_ack_claim_lost', {
+          eventId: event.id,
+          eventType,
+        });
+        return busyRetry();
+      }
+
       return NextResponse.json({ received: true }, { status: 200 });
     };
     // 1) Insert event idempotently (no transactions)
@@ -386,7 +467,23 @@ export async function POST(request: NextRequest) {
       }
       // processedAt is NULL => previous attempt failed; reprocess
     }
-
+    // P0.8: claim/lock BEFORE any business logic (multi-instance safe).
+    const claimState = await tryClaimStripeEvent(event.id);
+    if (claimState === 'already_processed') {
+      logInfo('stripe_webhook_duplicate_event', {
+        eventId: event.id,
+        eventType,
+        reason: 'already_processed',
+      });
+      return NextResponse.json({ received: true }, { status: 200 });
+    }
+    if (claimState === 'busy') {
+      logInfo('stripe_webhook_claimed_by_other_instance', {
+        eventId: event.id,
+        eventType,
+      });
+      return busyRetry();
+    }
     //2) Resolve orderId:
     //    primary: metadata.orderId
     //    fallback: orders.paymentIntentId == paymentIntentId (ONLY if unique match)
@@ -1123,6 +1220,21 @@ export async function POST(request: NextRequest) {
     }
 
     logError('Stripe webhook processing failed', error);
+    // P0.8: release claim early so Stripe retries can be claimed immediately.
+    try {
+      await db
+        .update(stripeEvents)
+        .set({ claimExpiresAt: new Date(0) })
+        .where(
+          and(
+            eq(stripeEvents.eventId, event.id),
+            eq(stripeEvents.claimedBy, STRIPE_WEBHOOK_INSTANCE_ID),
+            isNull(stripeEvents.processedAt)
+          )
+        );
+    } catch {
+      // best-effort
+    }
     return NextResponse.json({ error: 'internal_error' }, { status: 500 });
   }
 }

frontend/db/schema/shop.ts

@@ -269,10 +269,16 @@ export const stripeEvents = pgTable(
     orderId: uuid('order_id').references(() => orders.id),
     eventType: text('event_type').notNull(),
     paymentStatus: text('payment_status'),
+    claimedAt: timestamp('claimed_at', { withTimezone: true }),
+    claimExpiresAt: timestamp('claim_expires_at', { withTimezone: true }),
+    claimedBy: varchar('claimed_by', { length: 64 }),
     processedAt: timestamp('processed_at', { withTimezone: true }),
     createdAt: timestamp('created_at', { mode: 'date' }).defaultNow().notNull(),
   },
-  table => [uniqueIndex('stripe_events_event_id_idx').on(table.eventId)]
+  table => [
+    uniqueIndex('stripe_events_event_id_idx').on(table.eventId),
+    index('stripe_events_claim_expires_idx').on(table.claimExpiresAt),
+  ]
 );
 
 export const productPrices = pgTable(
@@ -405,6 +411,8 @@ export const paymentAttempts = pgTable(
     finalizedAt: timestamp('finalized_at', { withTimezone: true }),
   },
   t => [
+    check('payment_attempts_provider_check', sql`${t.provider} in ('stripe')`),
+
     // CHECKs (match SQL migration)
     check(
       'payment_attempts_status_check',

frontend/drizzle/0003_add_stripe_events_claim_lock.sql

@@ -0,0 +1,17 @@
+ALTER TABLE "stripe_events" ADD COLUMN "claimed_at" timestamp with time zone;--> statement-breakpoint
+ALTER TABLE "stripe_events" ADD COLUMN "claim_expires_at" timestamp with time zone;--> statement-breakpoint
+ALTER TABLE "stripe_events" ADD COLUMN "claimed_by" varchar(64);--> statement-breakpoint
+CREATE INDEX "stripe_events_claim_expires_idx" ON "stripe_events" USING btree ("claim_expires_at");--> statement-breakpoint
+
+DO $$
+BEGIN
+  IF NOT EXISTS (
+    SELECT 1
+    FROM pg_constraint
+    WHERE conname = 'payment_attempts_provider_check'
+  ) THEN
+    ALTER TABLE "payment_attempts"
+    ADD CONSTRAINT "payment_attempts_provider_check"
+    CHECK ("provider" in ('stripe'));
+  END IF;
+END $$;--> statement-breakpoint