fix(api): retry Play RTDN concurrency conflicts and transient web-push failures

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

@@ -5,6 +5,7 @@
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Orbit.Application.Behaviors;
 using Orbit.Application.Common;
 using Orbit.Application.Subscriptions.Services;
 using Orbit.Domain.Common;
@@ -14,7 +15,7 @@
 
 namespace Orbit.Application.Subscriptions.Commands;
 
-public record HandlePlayNotificationCommand(string PushBody) : IRequest<Result>;
+public record HandlePlayNotificationCommand(string PushBody) : IRequest<Result>, IConcurrencyRetryable;
 
 public partial class HandlePlayNotificationCommandHandler(
     IGenericRepository<User> userRepository,
@@ -27,7 +28,7 @@
 {
     private readonly GooglePlaySettings _settings = playSettings.Value;
 
     public async Task<Result> Handle(HandlePlayNotificationCommand request, CancellationToken cancellationToken)
     {
         DecodedPlayNotification? decoded;
         try
@@ -115,7 +116,7 @@
         {
             await unitOfWork.SaveChangesAsync(cancellationToken);
         }
-        catch (DbUpdateException) when (!string.IsNullOrEmpty(decoded.MessageId))
+        catch (DbUpdateException ex) when (ex is not DbUpdateConcurrencyException && !string.IsNullOrEmpty(decoded.MessageId))
         {
             if (await processedNotificationRepository.AnyAsync(p => p.MessageId == decoded.MessageId, cancellationToken))
             {

src/Orbit.Infrastructure/Services/PushNotificationService.cs

@@ -51,7 +51,7 @@
         }
     }
 
     private async Task SendFcm(
         List<Domain.Entities.PushSubscription> subs,
         string title, string body, string? url,
         List<Domain.Entities.PushSubscription> staleSubscriptions,
@@ -169,46 +169,74 @@
             await SendWebPushToSubscription(client, sub, message, staleSubscriptions, ct);
     }
 
     private async Task SendWebPushToSubscription(
         PushServiceClient client,
         Domain.Entities.PushSubscription sub,
         PushMessage message,
         List<Domain.Entities.PushSubscription> staleSubscriptions,
         CancellationToken ct)
     {
-        try
+        const int MaxRetries = 2;
+        const int BaseDelayMs = 300;
+
+        var pushSub = new Lib.Net.Http.WebPush.PushSubscription
+        {
+            Endpoint = sub.Endpoint,
+            Keys = new Dictionary<string, string>
+            {
+                ["p256dh"] = sub.P256dh,
+                ["auth"] = sub.Auth
+            }
+        };
+
+        for (var attempt = 0; ; attempt++)
         {
-            var pushSub = new Lib.Net.Http.WebPush.PushSubscription
+            try
             {
-                Endpoint = sub.Endpoint,
-                Keys = new Dictionary<string, string>
+                await client.RequestPushMessageDeliveryAsync(pushSub, message, ct);
+                if (logger.IsEnabled(LogLevel.Debug))
+                    LogWebPushSent(logger, sub.Endpoint);
+                return;
+            }
+            catch (PushServiceClientException ex)
+                when (ex.StatusCode == HttpStatusCode.Gone || ex.StatusCode == HttpStatusCode.NotFound)
+            {
+                if (logger.IsEnabled(LogLevel.Debug))
+                    LogWebPushSubscriptionGone(logger, sub.Endpoint);
+                staleSubscriptions.Add(sub);
+                return;
+            }
+            catch (PushServiceClientException ex)
+            {
+                if (!IsTransient(ex.StatusCode) || attempt >= MaxRetries)
                 {
-                    ["p256dh"] = sub.P256dh,
-                    ["auth"] = sub.Auth
+                    if (logger.IsEnabled(LogLevel.Warning))
+                        LogWebPushFailed(logger, sub.Endpoint, ex.StatusCode, ex.Message);
+                    return;
                 }
-            };
-            await client.RequestPushMessageDeliveryAsync(pushSub, message, ct);
-            if (logger.IsEnabled(LogLevel.Debug))
-                LogWebPushSent(logger, sub.Endpoint);
-        }
-        catch (PushServiceClientException ex)
-            when (ex.StatusCode == HttpStatusCode.Gone || ex.StatusCode == HttpStatusCode.NotFound)
-        {
-            if (logger.IsEnabled(LogLevel.Debug))
-                LogWebPushSubscriptionGone(logger, sub.Endpoint);
-            staleSubscriptions.Add(sub);
-        }
-        catch (PushServiceClientException ex)
-        {
-            if (logger.IsEnabled(LogLevel.Warning))
-                LogWebPushFailed(logger, sub.Endpoint, ex.StatusCode, ex.Message);
-        }
-        catch (Exception ex)
-        {
-            if (logger.IsEnabled(LogLevel.Warning))
-                LogWebPushFailedGeneric(logger, ex, sub.Endpoint);
+                await Task.Delay(BaseDelayMs << attempt, ct);
+            }
+            catch (Exception ex) when (!ct.IsCancellationRequested)
+            {
+                if (attempt >= MaxRetries)
+                {
+                    if (logger.IsEnabled(LogLevel.Warning))
+                        LogWebPushFailedGeneric(logger, ex, sub.Endpoint);
+                    return;
+                }
+                await Task.Delay(BaseDelayMs << attempt, ct);
+            }
         }
     }
+
+    private static bool IsTransient(HttpStatusCode? status) =>
+        status is null
+            or HttpStatusCode.RequestTimeout
+            or HttpStatusCode.TooManyRequests
+            or HttpStatusCode.InternalServerError
+            or HttpStatusCode.BadGateway
+            or HttpStatusCode.ServiceUnavailable
+            or HttpStatusCode.GatewayTimeout;
     [LoggerMessage(EventId = 1, Level = LogLevel.Warning, Message = "FCM is not initialized (Firebase credentials not configured). Skipping FCM push to {Count} subscription(s).")]
     private static partial void LogFcmNotInitialized(ILogger logger, int count);
 

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

@@ -2,14 +2,17 @@
 using System.Text;
 using System.Text.Json;
 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.Application.Subscriptions.Services;
+using Orbit.Domain.Common;
 using Orbit.Domain.Entities;
 using Orbit.Domain.Enums;
 using Orbit.Domain.Interfaces;
@@ -327,4 +330,49 @@ public async Task Handle_ReferralCoupon_SaveFails_CouponNotCancelled()
         await _referralConsumer.DidNotReceive().CancelConsumedCouponAsync(
             Arg.Any<Guid>(), Arg.Any<string>(), Arg.Any<CancellationToken>());
     }
+
+    [Fact]
+    public void Command_IsMarkedConcurrencyRetryable() =>
+        typeof(HandlePlayNotificationCommand).Should().BeAssignableTo<IConcurrencyRetryable>();
+
+    [Fact]
+    public async Task Handle_SaveThrowsConcurrencyConflict_PropagatesWithoutTreatingAsDuplicate()
+    {
+        var user = User.Create("Thomas", "test@example.com").Value;
+        StubUser(user);
+        StubVerify(new PlaySubscriptionState(true, DateTime.UtcNow.AddMonths(1), SubscriptionInterval.Monthly, true, "orbit_pro", null, null));
+        _unitOfWork.SaveChangesAsync(Arg.Any<CancellationToken>())
+            .ThrowsAsync(new DbUpdateConcurrencyException("stale xmin token"));
+
+        var act = () => _handler.Handle(new HandlePlayNotificationCommand(BuildPushBody(2, "tok_renew", "orbit_pro")), CancellationToken.None);
+
+        await act.Should().ThrowAsync<DbUpdateConcurrencyException>();
+        await _processedRepo.Received(1).AnyAsync(
+            Arg.Any<Expression<Func<ProcessedPlayNotification, bool>>>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task Handle_ThroughRetryBehavior_ConcurrencyConflictThenSuccess_GrantsProAndRetries()
+    {
+        var user = User.Create("Thomas", "test@example.com").Value;
+        StubUser(user);
+        StubVerify(new PlaySubscriptionState(true, DateTime.UtcNow.AddMonths(1), SubscriptionInterval.Monthly, true, "orbit_pro", null, null));
+
+        var saveAttempts = 0;
+        _unitOfWork.SaveChangesAsync(Arg.Any<CancellationToken>())
+            .Returns(_ => saveAttempts++ == 0
+                ? throw new DbUpdateConcurrencyException("stale xmin token")
+                : 1);
+
+        var command = new HandlePlayNotificationCommand(BuildPushBody(2, "tok_renew", "orbit_pro"));
+        var behavior = new ConcurrencyRetryBehavior<HandlePlayNotificationCommand, 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.PlayPurchaseToken.Should().Be("tok_renew");
+        _unitOfWork.Received(1).ResetTracking();
+    }
 }

tests/Orbit.Infrastructure.Tests/Services/PushNotificationServiceTests.cs

@@ -165,6 +165,67 @@ public async Task SendToUserAsync_MixedLiveAndDeadWebPush_PrunesOnlyDeadSubscrip
         remaining.Should().ContainSingle().Which.Should().Be(liveEndpoint);
     }
 
+    [Fact]
+    public async Task SendToUserAsync_WebPushTransientThenSuccess_RetriesAndKeepsSubscription()
+    {
+        await SeedWebPushSubscription("https://push.example.com/sub/retry");
+
+        var calls = 0;
+        var handler = new StubHttpMessageHandler(_ =>
+            new HttpResponseMessage(calls++ == 0 ? HttpStatusCode.ServiceUnavailable : HttpStatusCode.Created));
+        var service = CreateService(handler);
+
+        await service.SendToUserAsync(_userId, "Title", "Body");
+
+        handler.CallCount.Should().Be(2);
+        (await _dbContext.PushSubscriptions.CountAsync()).Should().Be(1);
+    }
+
+    [Fact]
+    public async Task SendToUserAsync_WebPushTransportErrorThenSuccess_RetriesAndKeepsSubscription()
+    {
+        await SeedWebPushSubscription("https://push.example.com/sub/transport");
+
+        var calls = 0;
+        var handler = new StubHttpMessageHandler(_ => calls++ == 0
+            ? throw new HttpRequestException("connection reset")
+            : new HttpResponseMessage(HttpStatusCode.Created));
+        var service = CreateService(handler);
+
+        await service.SendToUserAsync(_userId, "Title", "Body");
+
+        handler.CallCount.Should().Be(2);
+        (await _dbContext.PushSubscriptions.CountAsync()).Should().Be(1);
+    }
+
+    [Fact]
+    public async Task SendToUserAsync_WebPushSubscriptionGone_MarksStaleWithoutRetry()
+    {
+        await SeedWebPushSubscription("https://push.example.com/sub/gone");
+
+        var handler = new StubHttpMessageHandler(_ => new HttpResponseMessage(HttpStatusCode.Gone));
+        var service = CreateService(handler);
+
+        await service.SendToUserAsync(_userId, "Title", "Body");
+
+        handler.CallCount.Should().Be(1);
+        (await _dbContext.PushSubscriptions.CountAsync()).Should().Be(0);
+    }
+
+    [Fact]
+    public async Task SendToUserAsync_WebPushNonTransientStatus_DoesNotRetryOrPruneSubscription()
+    {
+        await SeedWebPushSubscription("https://push.example.com/sub/rejected");
+
+        var handler = new StubHttpMessageHandler(_ => new HttpResponseMessage(HttpStatusCode.BadRequest));
+        var service = CreateService(handler);
+
+        await service.SendToUserAsync(_userId, "Title", "Body");
+
+        handler.CallCount.Should().Be(1);
+        (await _dbContext.PushSubscriptions.CountAsync()).Should().Be(1);
+    }
+
     private static (string PublicKey, string PrivateKey) GenerateVapidKeyPair()
     {
         using var key = ECDsa.Create(ECCurve.NamedCurves.nistP256);