fix(api): retry Stripe-webhook concurrency conflicts + close idempotency TOCTOU (#243)

src/Orbit.Application/Subscriptions/Commands/HandleWebhookCommand.cs

@@ -1,6 +1,8 @@
 using MediatR;
+using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Orbit.Application.Behaviors;
 using Orbit.Application.Common;
 using Orbit.Domain.Common;
 using Orbit.Domain.Entities;
@@ -11,7 +13,7 @@
 
 namespace Orbit.Application.Subscriptions.Commands;
 
-public record HandleWebhookCommand(string Json, string Signature) : IRequest<Result>;
+public record HandleWebhookCommand(string Json, string Signature) : IRequest<Result>, IConcurrencyRetryable;
 
 public partial class HandleWebhookCommandHandler(
     IGenericRepository<User> userRepository,
@@ -50,12 +52,6 @@ public async Task<Result> Handle(HandleWebhookCommand request, CancellationToken
 
         LogStripeEventType(logger, stripeEvent.Type, stripeEvent.Id);
 
-        if (await processedEventRepository.AnyAsync(e => e.EventId == stripeEvent.Id, cancellationToken))
-        {
-            LogDuplicateEvent(logger, stripeEvent.Id);
-            return Result.Success();
-        }
-
         await processedEventRepository.AddAsync(ProcessedStripeEvent.Create(stripeEvent.Id), cancellationToken);
 
         try
@@ -89,7 +85,7 @@ public async Task<Result> Handle(HandleWebhookCommand request, CancellationToken
             LogErrorProcessingStripeEvent(logger, ex, stripeEvent.Type);
             return Result.Failure(ErrorMessages.WebhookStripeApiError);
         }
-        catch (Exception ex) when (ex is not OperationCanceledException)
+        catch (Exception ex) when (ex is not OperationCanceledException and not DbUpdateConcurrencyException)
         {
             LogErrorProcessingStripeEvent(logger, ex, stripeEvent.Type);
             return Result.Failure(ErrorMessages.WebhookProcessingFailed);

tests/Orbit.Application.Tests/Commands/Subscriptions/HandleWebhookCommandHandlerTests.cs

@@ -3,13 +3,16 @@
 using System.Security.Cryptography;
 using System.Text;
 using FluentAssertions;
+using MediatR;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using NSubstitute;
 using NSubstitute.ExceptionExtensions;
+using Orbit.Application.Behaviors;
 using Orbit.Application.Common;
 using Orbit.Application.Subscriptions.Commands;
+using Orbit.Domain.Common;
 using Orbit.Domain.Entities;
 using Orbit.Domain.Interfaces;
 using Stripe;
@@ -360,28 +363,34 @@ public async Task Handle_UnknownEventType_ReturnsSuccess()
     }
 
     [Fact]
-    public async Task Handle_DuplicateEventId_SkipsProcessingWithoutSaving()
+    public async Task Handle_ReservesProcessedEventViaConstraint_WithoutCheckThenInsertPreCheck()
     {
-        _processedRepo.AnyAsync(
-            Arg.Any<Expression<Func<ProcessedStripeEvent, bool>>>(),
+        var user = User.Create("Thomas", "test@example.com").Value;
+        _userRepo.FindOneTrackedIgnoringFiltersAsync(
+            Arg.Any<Expression<Func<User, bool>>>(),
             Arg.Any<CancellationToken>())
-            .Returns(true);
+            .Returns(user);
+
+        var subscription = CreateMockSubscription("sub_test");
+        _subscriptionService.GetAsync("sub_test", Arg.Any<SubscriptionGetOptions>(),
+            Arg.Any<RequestOptions>(), Arg.Any<CancellationToken>())
+            .Returns(subscription);
 
         var (json, signature) = BuildSignedEvent("checkout.session.completed", BuildCheckoutSessionJson(
             UserId.ToString(), "sub_test", "cus_test"));
 
-        var command = new HandleWebhookCommand(json, signature);
-        var result = await _handler.Handle(command, CancellationToken.None);
+        var result = await _handler.Handle(new HandleWebhookCommand(json, signature), CancellationToken.None);
 
         result.IsSuccess.Should().BeTrue();
-        await _userRepo.DidNotReceive().FindOneTrackedIgnoringFiltersAsync(
-            Arg.Any<Expression<Func<User, bool>>>(),
+        await _processedRepo.DidNotReceive().AnyAsync(
+            Arg.Any<Expression<Func<ProcessedStripeEvent, bool>>>(),
             Arg.Any<CancellationToken>());
-        await _unitOfWork.DidNotReceive().SaveChangesAsync(Arg.Any<CancellationToken>());
+        await _processedRepo.Received(1).AddAsync(
+            Arg.Any<ProcessedStripeEvent>(), Arg.Any<CancellationToken>());
     }
 
     [Fact]
-    public async Task Handle_ConcurrentDuplicate_SaveConflict_ReturnsSuccess()
+    public async Task Handle_DuplicateEvent_UniqueViolationOnSave_IsIdempotentSuccess()
     {
         var user = User.Create("Thomas", "test@example.com").Value;
         _userRepo.FindOneTrackedIgnoringFiltersAsync(
@@ -400,10 +409,72 @@ public async Task Handle_ConcurrentDuplicate_SaveConflict_ReturnsSuccess()
         var (json, signature) = BuildSignedEvent("checkout.session.completed", BuildCheckoutSessionJson(
             UserId.ToString(), "sub_test", "cus_test"));
 
+        var result = await _handler.Handle(new HandleWebhookCommand(json, signature), CancellationToken.None);
+
+        result.IsSuccess.Should().BeTrue();
+    }
+
+    [Fact]
+    public void Command_IsMarkedConcurrencyRetryable() =>
+        typeof(HandleWebhookCommand).Should().BeAssignableTo<IConcurrencyRetryable>();
+
+    [Fact]
+    public async Task Handle_SaveThrowsConcurrencyConflict_PropagatesInsteadOfSwallowingAsFailure()
+    {
+        var user = User.Create("Thomas", "test@example.com").Value;
+        _userRepo.FindOneTrackedIgnoringFiltersAsync(
+            Arg.Any<Expression<Func<User, bool>>>(),
+            Arg.Any<CancellationToken>())
+            .Returns(user);
+
+        var subscription = CreateMockSubscription("sub_test");
+        _subscriptionService.GetAsync("sub_test", Arg.Any<SubscriptionGetOptions>(),
+            Arg.Any<RequestOptions>(), Arg.Any<CancellationToken>())
+            .Returns(subscription);
+
+        _unitOfWork.SaveChangesAsync(Arg.Any<CancellationToken>())
+            .ThrowsAsync(new DbUpdateConcurrencyException("stale xmin token"));
+
+        var (json, signature) = BuildSignedEvent("checkout.session.completed", BuildCheckoutSessionJson(
+            UserId.ToString(), "sub_test", "cus_test"));
+
+        var act = () => _handler.Handle(new HandleWebhookCommand(json, signature), CancellationToken.None);
+
+        await act.Should().ThrowAsync<DbUpdateConcurrencyException>();
+    }
+
+    [Fact]
+    public async Task Handle_ThroughRetryBehavior_ConcurrencyConflictThenSuccess_UpgradesProAndRetries()
+    {
+        var user = User.Create("Thomas", "test@example.com").Value;
+        _userRepo.FindOneTrackedIgnoringFiltersAsync(
+            Arg.Any<Expression<Func<User, bool>>>(),
+            Arg.Any<CancellationToken>())
+            .Returns(user);
+
+        var subscription = CreateMockSubscription("sub_test");
+        _subscriptionService.GetAsync("sub_test", Arg.Any<SubscriptionGetOptions>(),
+            Arg.Any<RequestOptions>(), Arg.Any<CancellationToken>())
+            .Returns(subscription);
+
+        var saveAttempts = 0;
+        _unitOfWork.SaveChangesAsync(Arg.Any<CancellationToken>())
+            .Returns(_ => saveAttempts++ == 0
+                ? throw new DbUpdateConcurrencyException("stale xmin token")
+                : 1);
+
+        var (json, signature) = BuildSignedEvent("checkout.session.completed", BuildCheckoutSessionJson(
+            UserId.ToString(), "sub_test", "cus_test"));
         var command = new HandleWebhookCommand(json, signature);
-        var result = await _handler.Handle(command, CancellationToken.None);
+        var behavior = new ConcurrencyRetryBehavior<HandleWebhookCommand, Result>(_unitOfWork);
+        RequestHandlerDelegate<Result> next = cancellationToken => _handler.Handle(command, cancellationToken);
+
+        var result = await behavior.Handle(command, next, CancellationToken.None);
 
         result.IsSuccess.Should().BeTrue();
+        saveAttempts.Should().Be(2);
+        user.IsPro.Should().BeTrue();
+        _unitOfWork.Received(1).ResetTracking();
     }
 
     [Fact]