fix: Optimize cleanup jobs, eliminate raw ES client usage, add test coverage

src/Exceptionless.Core/Jobs/CleanupDataJob.cs

@@ -67,7 +67,7 @@ ILoggerFactory loggerFactory
 
     protected override Task<ILock?> GetLockAsync(CancellationToken cancellationToken = default)
     {
-        return _lockProvider.TryAcquireAsync(nameof(CleanupDataJob), TimeSpan.FromMinutes(15), cancellationToken);
+        return _lockProvider.TryAcquireAsync(nameof(CleanupDataJob), TimeSpan.FromHours(2), cancellationToken);
     }
 
     protected override async Task<JobResult> RunInternalAsync(JobContext context)
@@ -109,8 +109,13 @@ private async Task CleanupSoftDeletedOrganizationsAsync(JobContext context)
 
         while (organizationResults.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
+            await RenewLockAsync(context);
+
             foreach (var organization in organizationResults.Documents)
             {
+                if (context.CancellationToken.IsCancellationRequested)
+                    break;
+
                 using var _ = _logger.BeginScope(new ExceptionlessState().Organization(organization.Id));
                 try
                 {
@@ -137,8 +142,13 @@ private async Task CleanupSoftDeletedProjectsAsync(JobContext context)
 
         while (projectResults.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
+            await RenewLockAsync(context);
+
             foreach (var project in projectResults.Documents)
             {
+                if (context.CancellationToken.IsCancellationRequested)
+                    break;
+
                 using var _ = _logger.BeginScope(new ExceptionlessState().Organization(project.OrganizationId).Project(project.Id));
                 try
                 {
@@ -257,8 +267,13 @@ private async Task EnforceRetentionAsync(JobContext context)
         var results = await _organizationRepository.FindAsync(q => q.Include(o => o.Id, o => o.Name, o => o.RetentionDays), o => o.SearchAfterPaging().PageLimit(100));
         while (results.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested)
         {
+            await RenewLockAsync(context);
+
             foreach (var organization in results.Documents)
             {
+                if (context.CancellationToken.IsCancellationRequested)
+                    break;
+
                 using var _ = _logger.BeginScope(new ExceptionlessState().Organization(organization.Id));
 
                 int retentionDays = _billingManager.GetBillingPlanByUpsellingRetentionPeriod(organization.RetentionDays)?.RetentionDays ?? _appOptions.MaximumRetentionDays;

src/Exceptionless.Core/Repositories/EventRepository.cs

@@ -1,4 +1,5 @@
-using Exceptionless.Core.Models;
+using System.Linq.Expressions;
+using Exceptionless.Core.Models;
 using Exceptionless.Core.Repositories.Configuration;
 using Exceptionless.Core.Repositories.Queries;
 using Exceptionless.Core.Validation;
@@ -11,11 +12,13 @@ namespace Exceptionless.Core.Repositories;
 
 public class EventRepository : RepositoryOwnedByOrganizationAndProject<PersistentEvent>, IEventRepository
 {
+    private readonly ExceptionlessElasticConfiguration _configuration;
     private readonly TimeProvider _timeProvider;
 
     public EventRepository(ExceptionlessElasticConfiguration configuration, AppOptions options, MiniValidationValidator validator)
         : base(configuration.Events, validator, options)
     {
+        _configuration = configuration;
         _timeProvider = configuration.TimeProvider;
 
         DisableCache(); // NOTE: If cache is ever enabled, then fast paths for patching/deleting with scripts will be super slow!
@@ -189,9 +192,111 @@ public override Task<FindResults<PersistentEvent>> GetByProjectIdAsync(string pr
     public Task<long> RemoveAllByStackIdsAsync(string[] stackIds)
     {
         ArgumentNullException.ThrowIfNull(stackIds);
-        if (stackIds.Length == 0)
+        if (stackIds is [])
             throw new ArgumentOutOfRangeException(nameof(stackIds));
 
         return RemoveAllAsync(q => q.Stack(stackIds));
     }
+
+    public Task<long> RemoveAllByProjectIdsAsync(string[] projectIds)
+    {
+        ArgumentNullException.ThrowIfNull(projectIds);
+        if (projectIds is [])
+            throw new ArgumentOutOfRangeException(nameof(projectIds));
+
+        return RemoveAllAsync(q => q.Project(projectIds));
+    }
+
+    public Task<long> RemoveAllByOrganizationIdsAsync(string[] organizationIds)
+    {
+        ArgumentNullException.ThrowIfNull(organizationIds);
+        if (organizationIds is [])
+            throw new ArgumentOutOfRangeException(nameof(organizationIds));
+
+        return RemoveAllAsync(q => q.Organization(organizationIds));
+    }
+
+    /// <summary>
+    /// Reassigns all events from the source stacks to the target stack using a parameterized
+    /// Painless script (no string interpolation) to prevent script injection.
+    /// </summary>
+    public Task<long> ReassignStackAsync(IEnumerable<string> sourceStackIds, string targetStackId)
+    {
+        ArgumentNullException.ThrowIfNull(sourceStackIds);
+        ArgumentException.ThrowIfNullOrEmpty(targetStackId);
+
+        // Materialize to avoid multiple enumeration and guard against empty — an empty
+        // .Stack() filter would match ALL events and reassign them to the target stack.
+        var sourceIds = sourceStackIds.ToList();
+        if (sourceIds.Count == 0)
+            return Task.FromResult(0L);
+
+        return PatchAllAsync(
+            q => q.Stack(sourceIds),
+            new ScriptPatch("ctx._source.stack_id = params.targetStackId")
+            {
+                Params = new Dictionary<string, object> { ["targetStackId"] = targetStackId }
+            });
+    }
+
+    public Task<IReadOnlyCollection<string>> GetDistinctStackIdsAsync(int batchSize, CompositeKeyResult? afterKey = null)
+    {
+        return GetDistinctFieldValuesAsync("stack_id", e => e.StackId, batchSize, afterKey);
+    }
+
+    public Task<IReadOnlyCollection<string>> GetDistinctProjectIdsAsync(int batchSize, CompositeKeyResult? afterKey = null)
+    {
+        return GetDistinctFieldValuesAsync("project_id", e => e.ProjectId, batchSize, afterKey);
+    }
+
+    public Task<IReadOnlyCollection<string>> GetDistinctOrganizationIdsAsync(int batchSize, CompositeKeyResult? afterKey = null)
+    {
+        return GetDistinctFieldValuesAsync("organization_id", e => e.OrganizationId, batchSize, afterKey);
+    }
+
+    /// <summary>
+    /// Uses a composite aggregation to paginate through all distinct values of a field.
+    /// Composite aggregations are preferred over terms aggregations for high-cardinality fields
+    /// because terms aggregations can silently miss values when the unique count exceeds the
+    /// configured size parameter. Composite aggregations guarantee correct iteration via an
+    /// after_key cursor, at the cost of requiring sequential page fetches.
+    /// </summary>
+    private async Task<IReadOnlyCollection<string>> GetDistinctFieldValuesAsync(
+        string fieldName, Expression<Func<PersistentEvent, object>> fieldExpression, int batchSize, CompositeKeyResult? afterKey)
+    {
+        var afterKeyValues = afterKey?.AfterKey;
+        var search = await _configuration.Client.SearchAsync<PersistentEvent>(s =>
+        {
+            s.Size(0).Aggregations(a => a
+                .Composite($"composite_{fieldName}", c =>
+                {
+                    var composite = c.Size(batchSize)
+                        .Sources(src => src.Terms(fieldName, t => t.Field(fieldExpression)));
+                    if (afterKeyValues is { Count: > 0 })
+                        composite.After(new CompositeKey(afterKeyValues));
+                    return composite;
+                }));
+            return s;
+        });
+
+        var composite = search.Aggregations.Composite($"composite_{fieldName}");
+
+        // Always clear the cursor first; repopulate only when a next page exists.
+        // This ensures callers that check afterKey.AfterKey.Count > 0 correctly
+        // detect end-of-pagination without requiring a final empty-result fetch.
+        if (afterKey is not null)
+        {
+            afterKey.AfterKey.Clear();
+            if (composite?.AfterKey is not null)
+            {
+                foreach (var kvp in composite.AfterKey)
+                    afterKey.AfterKey[kvp.Key] = kvp.Value;
+            }
+        }
+
+        if (composite?.Buckets is not { Count: > 0 })
+            return [];
+
+        return composite.Buckets.Select(b => b.Key[fieldName].ToString()!).ToArray();
+    }
 }

src/Exceptionless.Core/Repositories/Interfaces/IEventRepository.cs

@@ -13,6 +13,17 @@ public interface IEventRepository : IRepositoryOwnedByOrganizationAndProject<Per
     Task<bool> UpdateSessionStartLastActivityAsync(string id, DateTime lastActivityUtc, bool isSessionEnd = false, bool hasError = false, bool sendNotifications = true);
     Task<long> RemoveAllAsync(string organizationId, string? clientIpAddress, DateTime? utcStart, DateTime? utcEnd, CommandOptionsDescriptor<PersistentEvent>? options = null);
     Task<long> RemoveAllByStackIdsAsync(string[] stackIds);
+    Task<long> RemoveAllByProjectIdsAsync(string[] projectIds);
+    Task<long> RemoveAllByOrganizationIdsAsync(string[] organizationIds);
+    Task<long> ReassignStackAsync(IEnumerable<string> sourceStackIds, string targetStackId);
+    Task<IReadOnlyCollection<string>> GetDistinctStackIdsAsync(int batchSize, CompositeKeyResult? afterKey = null);
+    Task<IReadOnlyCollection<string>> GetDistinctProjectIdsAsync(int batchSize, CompositeKeyResult? afterKey = null);
+    Task<IReadOnlyCollection<string>> GetDistinctOrganizationIdsAsync(int batchSize, CompositeKeyResult? afterKey = null);
+}
+
+public record CompositeKeyResult
+{
+    public Dictionary<string, object> AfterKey { get; init; } = [];
 }
 
 public static class EventRepositoryExtensions

src/Exceptionless.Core/Repositories/Interfaces/IStackRepository.cs

@@ -15,4 +15,5 @@ public interface IStackRepository : IRepositoryOwnedByOrganizationAndProject<Sta
     Task<FindResults<Stack>> GetStacksForCleanupAsync(string organizationId, DateTime cutoff);
     Task<FindResults<Stack>> GetSoftDeleted();
     Task<long> SoftDeleteByProjectIdAsync(string organizationId, string projectId);
+    Task<IReadOnlyCollection<string>> GetDuplicateSignaturesAsync(int maxResults = 10000);
 }

src/Exceptionless.Core/Repositories/StackRepository.cs

@@ -188,6 +188,22 @@ public Task<long> SoftDeleteByProjectIdAsync(string organizationId, string proje
         );
     }
 
+    public async Task<IReadOnlyCollection<string>> GetDuplicateSignaturesAsync(int maxResults = 10000)
+    {
+        // ImmediateConsistency forces a segment refresh before the aggregation so that
+        // any stacks soft-deleted in a previous batch are excluded here. Cost: one refresh
+        // per batch (not per item), equivalent to the original explicit index refresh.
+        var result = await CountAsync(
+            q => q.AggregationsExpression($"terms:(duplicate_signature~{maxResults} @min:2)"),
+            o => o.ImmediateConsistency());
+
+        var buckets = result.Aggregations.Terms("terms_duplicate_signature")?.Buckets;
+        if (buckets is not { Count: > 0 })
+            return [];
+
+        return buckets.Select(b => b.Key).ToArray();
+    }
+
     protected override async Task AddDocumentsToCacheAsync(ICollection<FindHit<Stack>> findHits, ICommandOptions options, bool isDirtyRead)
     {
         await base.AddDocumentsToCacheAsync(findHits, options, isDirtyRead);