perf(api): parallelize/batch background fan-out to cut sequential latency

src/Orbit.Application/Challenges/Commands/CreateChallengeCommand.cs

@@ -1,5 +1,6 @@
 using System.Security.Cryptography;
 using MediatR;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Orbit.Application.Common;
 using Orbit.Application.Social.Services;
@@ -21,18 +22,19 @@
     IReadOnlyList<Guid> LinkedHabitIds,
     IReadOnlyList<Guid> InvitedFriendUserIds) : IRequest<Result<Guid>>;
 
 public partial class CreateChallengeCommandHandler(
     SocialAccessGuard socialAccessGuard,
     FriendGraphService friendGraphService,
     IGenericRepository<Challenge> challengeRepository,
     IGenericRepository<Habit> habitRepository,
     IGenericRepository<User> userRepository,
-    IPushNotificationService pushNotificationService,
+    IServiceScopeFactory scopeFactory,
     IUnitOfWork unitOfWork,
     ILogger<CreateChallengeCommandHandler> logger) : IRequestHandler<CreateChallengeCommand, Result<Guid>>
 {
     private const string JoinCodeAlphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789";
     private const int JoinCodeLength = 8;
+    private const int MaxInvitePushConcurrency = 5;
 
     public async Task<Result<Guid>> Handle(CreateChallengeCommand request, CancellationToken cancellationToken)
     {
@@ -90,22 +92,33 @@
             return;
 
         var invitedUsers = await userRepository.FindAsync(u => invitedFriendIds.Contains(u.Id), cancellationToken);
-        foreach (var user in invitedUsers)
+
+        using var throttle = new SemaphoreSlim(MaxInvitePushConcurrency);
+        await Task.WhenAll(invitedUsers.Select(user => NotifyOneAsync(requester, user, throttle, cancellationToken)));
+    }
+
+    private async Task NotifyOneAsync(User requester, User user, SemaphoreSlim throttle, CancellationToken cancellationToken)
+    {
+        await throttle.WaitAsync(cancellationToken);
+        try
         {
             var isPortuguese = LocaleHelper.IsPortuguese(user.Language);
             var title = isPortuguese ? "Novo desafio em grupo" : "New group challenge";
             var body = isPortuguese
                 ? $"{requester.Name} convidou você para um desafio."
                 : $"{requester.Name} invited you to a challenge.";
 
-            try
-            {
-                await pushNotificationService.SendToUserAsync(user.Id, title, body, cancellationToken: cancellationToken);
-            }
-            catch (Exception ex)
-            {
-                LogPushNotificationFailed(logger, ex, user.Id);
-            }
+            using var scope = scopeFactory.CreateScope();
+            var pushNotificationService = scope.ServiceProvider.GetRequiredService<IPushNotificationService>();
+            await pushNotificationService.SendToUserAsync(user.Id, title, body, cancellationToken: cancellationToken);
+        }
+        catch (Exception ex)
+        {
+            LogPushNotificationFailed(logger, ex, user.Id);
+        }
+        finally
+        {
+            throttle.Release();
         }
     }
 

src/Orbit.Infrastructure/Services/ReminderSchedulerService.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
@@ -21,6 +22,9 @@ public partial class ReminderSchedulerService(
     private readonly TimeSpan _interval = TimeSpan.FromMinutes(
         configuration.GetValue("BackgroundServices:ReminderIntervalMinutes", 1));
 
+    private readonly int _pushConcurrency = Math.Max(1,
+        configuration.GetValue("BackgroundServices:ReminderPushConcurrency", 8));
+
     public string Name => "reminder-scheduler";
 
     public string CronExpression => "* * * * *";
@@ -45,14 +49,17 @@ internal async Task CheckAndSendReminders(CancellationToken ct)
     {
         using var scope = scopeFactory.CreateScope();
         var dbContext = scope.ServiceProvider.GetRequiredService<OrbitDbContext>();
-        var pushService = scope.ServiceProvider.GetRequiredService<IPushNotificationService>();
 
-        await ProcessRelativeReminders(dbContext, pushService, ct);
+        var pending = new List<PendingReminderPush>();
+
+        await ProcessRelativeReminders(dbContext, pending, ct);
 
-        await ProcessScheduledReminders(dbContext, pushService, ct);
+        await ProcessScheduledReminders(dbContext, pending, ct);
+
+        await SendPushesAsync(pending, ct);
     }
 
-    private async Task ProcessRelativeReminders(OrbitDbContext dbContext, IPushNotificationService pushService, CancellationToken ct)
+    private async Task ProcessRelativeReminders(OrbitDbContext dbContext, List<PendingReminderPush> pending, CancellationToken ct)
     {
         var minLocalDate = DateOnly.FromDateTime(DateTime.UtcNow.AddDays(-1));
         var maxLocalDate = DateOnly.FromDateTime(DateTime.UtcNow.AddDays(1));
@@ -98,15 +105,15 @@ private async Task ProcessRelativeReminders(OrbitDbContext dbContext, IPushNotif
         foreach (var habit in habits)
         {
             await ProcessSingleRelativeReminderAsync(
-                habit, users, loggedHabitDates, sentReminderSet, pushService, dbContext, ct);
+                habit, users, loggedHabitDates, sentReminderSet, pending, dbContext, ct);
         }
     }
 
     private async Task ProcessSingleRelativeReminderAsync(
         Habit habit, Dictionary<Guid, User> users,
         HashSet<(Guid HabitId, DateOnly Date)> loggedHabitDates,
         HashSet<(Guid HabitId, DateOnly Date, int MinutesBefore)> sentReminderSet,
-        IPushNotificationService pushService, OrbitDbContext dbContext, CancellationToken ct)
+        List<PendingReminderPush> pending, OrbitDbContext dbContext, CancellationToken ct)
     {
         if (!users.TryGetValue(habit.UserId, out var user)) return;
 
@@ -132,15 +139,17 @@ private async Task ProcessSingleRelativeReminderAsync(
 
             sentReminderSet.Add((habit.Id, userToday, minutesBefore));
 
-            if (!await TryRecordAndSendAsync(habit, sentReminder, notification, minutesText, pushService, dbContext, ct))
+            if (!await TryRecordReminderAsync(habit, sentReminder, notification, dbContext, ct))
                 continue;
 
+            pending.Add(new PendingReminderPush(habit.UserId, habit.Title, minutesText, habit.Id));
+
             if (logger.IsEnabled(LogLevel.Debug))
                 LogSentReminder(logger, minutesBefore, habit.Id, habit.UserId);
         }
     }
 
-    private async Task ProcessScheduledReminders(OrbitDbContext dbContext, IPushNotificationService pushService, CancellationToken ct)
+    private async Task ProcessScheduledReminders(OrbitDbContext dbContext, List<PendingReminderPush> pending, CancellationToken ct)
     {
         var minLocalDate = DateOnly.FromDateTime(DateTime.UtcNow.AddDays(-1));
         var maxDayBeforeDate = DateOnly.FromDateTime(DateTime.UtcNow.AddDays(2));
@@ -177,14 +186,14 @@ private async Task ProcessScheduledReminders(OrbitDbContext dbContext, IPushNoti
         foreach (var habit in habits)
         {
             await ProcessSingleScheduledReminderAsync(
-                habit, users, sentScheduledSet, pushService, dbContext, ct);
+                habit, users, sentScheduledSet, pending, dbContext, ct);
         }
     }
 
     private async Task ProcessSingleScheduledReminderAsync(
         Habit habit, Dictionary<Guid, User> users,
         HashSet<(Guid HabitId, DateOnly Date, TimeOnly ReminderTimeUtc, ScheduledReminderWhen? When)> sentScheduledSet,
-        IPushNotificationService pushService, OrbitDbContext dbContext, CancellationToken ct)
+        List<PendingReminderPush> pending, OrbitDbContext dbContext, CancellationToken ct)
     {
         if (!users.TryGetValue(habit.UserId, out var user)) return;
 
@@ -213,9 +222,11 @@ private async Task ProcessSingleScheduledReminderAsync(
 
             sentScheduledSet.Add((habit.Id, userToday, sr.Time, sr.When));
 
-            if (!await TryRecordAndSendAsync(habit, sentReminder, notification, text, pushService, dbContext, ct))
+            if (!await TryRecordReminderAsync(habit, sentReminder, notification, dbContext, ct))
                 continue;
 
+            pending.Add(new PendingReminderPush(habit.UserId, habit.Title, text, habit.Id));
+
             if (logger.IsEnabled(LogLevel.Debug))
                 LogSentScheduledReminder(logger, sr.When, sr.Time, habit.Id, habit.UserId);
         }
@@ -231,9 +242,9 @@ private static bool ShouldSendScheduledReminder(
         return userTimeNow >= sr.Time;
     }
 
-    private async Task<bool> TryRecordAndSendAsync(
-        Habit habit, SentReminder sentReminder, Notification notification, string body,
-        IPushNotificationService pushService, OrbitDbContext dbContext, CancellationToken ct)
+    private async Task<bool> TryRecordReminderAsync(
+        Habit habit, SentReminder sentReminder, Notification notification,
+        OrbitDbContext dbContext, CancellationToken ct)
     {
         await dbContext.SentReminders.AddAsync(sentReminder, ct);
         await dbContext.Notifications.AddAsync(notification, ct);
@@ -256,16 +267,48 @@ private async Task<bool> TryRecordAndSendAsync(
             return false;
         }
 
-        await pushService.SendToUserAsync(habit.UserId, habit.Title, body, "/", ct);
         return true;
     }
 
+    private async Task SendPushesAsync(IReadOnlyList<PendingReminderPush> pending, CancellationToken ct)
+    {
+        if (pending.Count == 0) return;
+
+        var concurrency = Math.Min(_pushConcurrency, pending.Count);
+        var queue = new ConcurrentQueue<PendingReminderPush>(pending);
+        var workers = new Task[concurrency];
+        for (var i = 0; i < concurrency; i++)
+            workers[i] = DrainPushQueueAsync(queue, ct);
+
+        await Task.WhenAll(workers);
+    }
+
+    private async Task DrainPushQueueAsync(ConcurrentQueue<PendingReminderPush> queue, CancellationToken ct)
+    {
+        using var scope = scopeFactory.CreateScope();
+        var pushService = scope.ServiceProvider.GetRequiredService<IPushNotificationService>();
+
+        while (queue.TryDequeue(out var push))
+        {
+            try
+            {
+                await pushService.SendToUserAsync(push.UserId, push.Title, push.Body, "/", ct);
+            }
+            catch (Exception ex) when (ex is not OperationCanceledException)
+            {
+                LogReminderPushFailed(logger, push.HabitId, push.UserId, ex);
+            }
+        }
+    }
+
     private static void DetachPendingEntries(OrbitDbContext dbContext)
     {
         foreach (var entry in dbContext.ChangeTracker.Entries().ToList())
             entry.State = EntityState.Detached;
     }
 
+    private readonly record struct PendingReminderPush(Guid UserId, string Title, string Body, Guid HabitId);
+
     private static string Pluralize(string singular, int count) => count > 1 ? singular + "s" : singular;
 
     private static string FormatReminderText(int minutesBefore, string lang)
@@ -318,4 +361,7 @@ private static string FormatScheduledReminderText(ScheduledReminderWhen when, st
     [LoggerMessage(EventId = 7, Level = LogLevel.Error, Message = "Failed to record reminder for habit {HabitId} (user {UserId}); skipping push")]
     private static partial void LogReminderRecordFailed(ILogger logger, Guid habitId, Guid userId, Exception ex);
 
+    [LoggerMessage(EventId = 8, Level = LogLevel.Warning, Message = "Failed to deliver reminder push for habit {HabitId} (user {UserId})")]
+    private static partial void LogReminderPushFailed(ILogger logger, Guid habitId, Guid userId, Exception ex);
+
 }