fix(referrals): Reprocess pending usages when program completion requirements are reduced (#1737)

* fix(referrals): Reprocess pending usages when program completion requirements are reduced

* Add batched and throttled parallel processing for usage progress and expiration

Author: adrianwium

src/api/src/domain/Yoma.Core.Domain/Referral/Enumerations.cs

@@ -164,6 +164,7 @@ public enum ReferralBlockReason
 
   public enum ReferralTriggerSource
   {
+    ProgramUpdated,
     IdentityAction,
     OpportunityCompletion
   }

src/api/src/domain/Yoma.Core.Domain/Referral/Events/ReferralProgressTriggerEventHandler.cs

@@ -28,10 +28,13 @@ public async Task Handle(ReferralProgressTriggerEvent notification, Cancellation
       {
         switch (notification.Entity.Source)
         {
+          case ReferralTriggerSource.ProgramUpdated:
           case ReferralTriggerSource.IdentityAction:
-            if (_logger.IsEnabled(LogLevel.Information)) _logger.LogInformation("Referral progress: handling identity trigger for user {UserId} ({Username})",
-              notification.Entity.UserId,
-              notification.Entity.Username);
+            if (_logger.IsEnabled(LogLevel.Information))
+              _logger.LogInformation("Referral progress: handling {Source} trigger for user {UserId} ({Username})",
+                notification.Entity.Source,
+                notification.Entity.UserId,
+                notification.Entity.Username);
             break;
 
           case ReferralTriggerSource.OpportunityCompletion:
@@ -42,7 +45,8 @@ public async Task Handle(ReferralProgressTriggerEvent notification, Cancellation
               throw new InvalidOperationException("OpportunityTitle must be provided for OpportunityCompletion source");
 
             if (_logger.IsEnabled(LogLevel.Information)) _logger.LogInformation(
-              "Referral progress: handling Opportunity trigger for user {UserId} ({Username}) — opportunity {OpportunityId} '{OpportunityTitle}'",
+              "Referral progress: handling {Source} trigger for user {UserId} ({Username}) — opportunity {OpportunityId} '{OpportunityTitle}'",
+              notification.Entity.Source,
               notification.Entity.UserId,
               notification.Entity.Username,
               notification.Entity.OpportunityId,
@@ -59,7 +63,7 @@ public async Task Handle(ReferralProgressTriggerEvent notification, Cancellation
       {
         if (_logger.IsEnabled(LogLevel.Error)) _logger.LogError(ex,
           "Referral progress: failed processing {Source} trigger for user {UserId} ({Username}) — error: {ErrorMessage}",
-          notification.Entity.Source.ToString(),
+          notification.Entity.Source,
           notification.Entity.UserId,
           notification.Entity.Username,
           ex.Message);

src/api/src/domain/Yoma.Core.Domain/Referral/Interfaces/ILinkMaintenanceService.cs

@@ -18,5 +18,7 @@ public interface ILinkMaintenanceService
     Task CancelByProgramId(List<Guid> programIds, ILogger? logger = null);
 
     Task ExpireByProgramId(List<Guid> programIds, ILogger? logger = null);
+
+    Task ProcessUsageProgressByProgramId(Guid programId, ILogger? logger = null);
   }
 }

src/api/src/domain/Yoma.Core.Domain/Referral/Services/LinkMaintenanceService.cs

@@ -1,7 +1,9 @@
+using MediatR;
 using Microsoft.Extensions.Logging;
 using Yoma.Core.Domain.Core.Exceptions;
 using Yoma.Core.Domain.Core.Helpers;
 using Yoma.Core.Domain.Core.Interfaces;
+using Yoma.Core.Domain.Referral.Events;
 using Yoma.Core.Domain.Referral.Interfaces;
 using Yoma.Core.Domain.Referral.Interfaces.Lookups;
 using Yoma.Core.Domain.Referral.Models;
@@ -14,17 +16,22 @@ public class LinkMaintenanceService : ILinkMaintenanceService
     private readonly ILinkStatusService _linkStatusService;
     private readonly ILinkUsageStatusService _linkUsageStatusService;
 
+    private readonly IMediator _mediator;
     private readonly IExecutionStrategyService _executionStrategyService;
 
     private readonly IRepositoryBatchedValueContainsWithNavigation<ReferralLink> _linkRepository;
     private readonly IRepositoryBatched<ReferralLinkUsage> _linkUsageRepository;
+
+    private const int Processing_BatchSize = 1000;
+    private const int Processing_Parallelism = 10;
     #endregion
 
     #region Constructor
     public LinkMaintenanceService(
       ILinkStatusService linkStatusService,
       ILinkUsageStatusService linkUsageStatusService,
 
+      IMediator mediator,
       IExecutionStrategyService executionStrategyService,
 
       IRepositoryBatchedValueContainsWithNavigation<ReferralLink> linkRepository,
@@ -33,6 +40,7 @@ public LinkMaintenanceService(
       _linkStatusService = linkStatusService ?? throw new ArgumentNullException(nameof(linkStatusService));
       _linkUsageStatusService = linkUsageStatusService ?? throw new ArgumentNullException(nameof(linkUsageStatusService));
 
+      _mediator = mediator ?? throw new ArgumentNullException(nameof(mediator));
       _executionStrategyService = executionStrategyService ?? throw new ArgumentNullException(nameof(executionStrategyService));
 
       _linkRepository = linkRepository ?? throw new ArgumentNullException(nameof(linkRepository));
@@ -214,6 +222,61 @@ await _executionStrategyService.ExecuteInExecutionStrategyAsync(async () =>
 
       logger.LogInformation("Expired {Total} link(s) across {ProgramCount} program(s)", items.Count, byProgram.Count);
     }
+
+    /// <summary>
+    /// Trigger sweep: reprocesses all pending link usages for a program when completion requirements are reduced
+    /// (e.g. Proof of Personhood or Pathway removed), ensuring users are re-evaluated against updated rules.
+    /// </summary>
+    public async Task ProcessUsageProgressByProgramId(Guid programId, ILogger? logger = null)
+    {
+      if (programId == Guid.Empty) throw new ArgumentNullException(nameof(programId));
+
+      var statusPendingId = _linkUsageStatusService.GetByName(ReferralLinkUsageStatus.Pending.ToString()).Id;
+      var totalProcessed = 0;
+      var page = 0;
+
+      using var throttler = new SemaphoreSlim(Processing_Parallelism, Processing_Parallelism);
+
+      while (true)
+      {
+        var items = _linkUsageRepository.Query()
+          .Where(o => o.ProgramId == programId && o.StatusId == statusPendingId)
+          .OrderBy(o => o.Id)
+          .Take(Processing_BatchSize)
+          .Select(o => new { o.UserId, o.Username, o.UserDisplayName })
+          .ToList();
+
+        if (items.Count == 0) break;
+
+        var tasks = items.Select(async item =>
+        {
+          await throttler.WaitAsync();
+          try
+          {
+            await _mediator.Publish(new ReferralProgressTriggerEvent(new ReferralProgressTriggerMessage
+            {
+              Source = ReferralTriggerSource.ProgramUpdated,
+              UserId = item.UserId,
+              Username = item.Username,
+              UserDisplayName = item.UserDisplayName
+            }));
+          }
+          finally { throttler.Release(); }
+        });
+
+        await Task.WhenAll(tasks);
+
+        totalProcessed += items.Count;
+        page++;
+      }
+
+      if (logger == null || !logger.IsEnabled(LogLevel.Information)) return;
+
+      if (totalProcessed == 0)
+        logger.LogInformation("No pending link usages found for program {ProgramId}", programId);
+      else
+        logger.LogInformation("Processed {Count} pending link usage(s) for program {ProgramId}", totalProcessed, programId);
+    }
     #endregion
 
     #region Private Members
@@ -224,25 +287,36 @@ private async Task ExpireLinkUsagesByLinkId(List<Guid> linkIds, ILogger? logger
     {
       if (linkIds == null || linkIds.Count == 0 || linkIds.Any(o => o == Guid.Empty))
         throw new ArgumentNullException(nameof(linkIds));
+
       linkIds = [.. linkIds.Distinct()];
 
       var statusExpiredId = _linkUsageStatusService.GetByName(ReferralLinkUsageStatus.Expired.ToString()).Id;
       var statusExpirableIds = LinkUsageBackgroundService.Statuses_Expirable.Select(o => _linkUsageStatusService.GetByName(o.ToString()).Id).ToList();
 
-      var items = _linkUsageRepository.Query()
-        .Where(o => linkIds.Contains(o.LinkId) && statusExpirableIds.Contains(o.StatusId))
-        .ToList();
+      var totalProcessed = 0;
 
-      if (items.Count == 0)
+      while (true)
       {
-        if (logger?.IsEnabled(LogLevel.Information) == true) logger.LogInformation("No expirable link usages found for {LinkCount} link(s)", linkIds.Count);
-        return;
+        var items = _linkUsageRepository.Query()
+          .Where(o => linkIds.Contains(o.LinkId) && statusExpirableIds.Contains(o.StatusId))
+          .OrderBy(o => o.Id)
+          .Take(Processing_BatchSize)
+          .ToList();
+
+        if (items.Count == 0) break;
+
+        items.ForEach(o => { o.StatusId = statusExpiredId; o.Status = ReferralLinkUsageStatus.Expired; });
+        await _linkUsageRepository.Update(items);
+
+        totalProcessed += items.Count;
       }
 
-      items.ForEach(o => { o.StatusId = statusExpiredId; o.Status = ReferralLinkUsageStatus.Expired; });
-      await _linkUsageRepository.Update(items);
+      if (logger == null || !logger.IsEnabled(LogLevel.Information)) return;
 
-      if (logger?.IsEnabled(LogLevel.Information) == true) logger.LogInformation("Expired {Count} link usage(s) across {LinkCount} link(s)", items.Count, linkIds.Count);
+      if (totalProcessed == 0)
+        logger.LogInformation("No expirable link usages found for {LinkCount} link(s)", linkIds.Count);
+      else
+        logger.LogInformation("Expired {Count} link usage(s) across {LinkCount} link(s)", totalProcessed, linkIds.Count);
     }
     #endregion
   }

src/api/src/domain/Yoma.Core.Domain/Referral/Services/ProgramService.cs

@@ -510,6 +510,9 @@ public async Task<Program> Update(ProgramRequestUpdate request)
 
       var user = _userService.GetByUsername(HttpContextAccessorHelper.GetUsername(_httpContextAccessor, false), false, false);
 
+      var pathwayRemoved = result.Pathway != null && request.Pathway == null;
+      var popRemoved = result.ProofOfPersonhoodRequired && !request.ProofOfPersonhoodRequired;
+
       result.Name = request.Name;
       result.Summary = request.Summary;
       result.Description = request.Description;
@@ -568,6 +571,11 @@ await _executionStrategyService.ExecuteInExecutionStrategyAsync(async () =>
 
       if (_logger.IsEnabled(LogLevel.Information)) _logger.LogInformation("Program {ProgramId} updated. FinalStatus={Status}", result.Id, result.Status);
 
+      // Trigger sweep: reprocess pending usages when completion requirements are reduced (POP / Pathway removed)
+      if (result.Status == ProgramStatus.Active && (popRemoved || pathwayRemoved))
+        if (result.Status == ProgramStatus.Active && (popRemoved || pathwayRemoved))
+          await _linkMaintenanceService.ProcessUsageProgressByProgramId(result.Id, _logger);
+
       return result;
     }