Support completion persistence and job name filtering in the background jobs module

modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs

@@ -48,6 +48,12 @@ public class BackgroundJobRecord : AggregateRoot<Guid>, IHasCreationTime
     /// </summary>
     public virtual bool IsAbandoned { get; set; }
 
+    /// <summary>
+    /// The time this job was completed successfully. When set, the job is kept as history and excluded
+    /// from the waiting jobs query (set only when successful job persistence is enabled).
+    /// </summary>
+    public virtual DateTime? CompletionTime { get; set; }
+
     /// <summary>
     /// Priority of this job.
     /// </summary>

modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.ObjectMapping;
@@ -22,9 +23,13 @@ public class BackgroundJobStore : IBackgroundJobStore, ITransientDependency
 
     public virtual async Task<BackgroundJobInfo> FindAsync(Guid jobId)
     {
-        return ObjectMapper.Map<BackgroundJobRecord, BackgroundJobInfo>(
-            await BackgroundJobRepository.FindAsync(jobId)
-        );
+        var backgroundJobRecord = await BackgroundJobRepository.FindAsync(jobId);
+        if (backgroundJobRecord == null)
+        {
+            return null!;
+        }
+
+        return ObjectMapper.Map<BackgroundJobRecord, BackgroundJobInfo>(backgroundJobRecord);
     }
 
     public virtual async Task InsertAsync(BackgroundJobInfo jobInfo)
@@ -34,18 +39,37 @@ public class BackgroundJobStore : IBackgroundJobStore, ITransientDependency
         );
     }
 
-    public virtual async Task<List<BackgroundJobInfo>> GetWaitingJobsAsync(string applicationName, int maxResultCount)
+    public virtual async Task<List<BackgroundJobInfo>> GetWaitingJobsAsync(string? applicationName, int maxResultCount)
     {
         return ObjectMapper.Map<List<BackgroundJobRecord>, List<BackgroundJobInfo>>(
             await BackgroundJobRepository.GetWaitingListAsync(applicationName, maxResultCount)
         );
     }
 
+    public virtual async Task<List<BackgroundJobInfo>> GetWaitingJobsAsync(
+        string? applicationName,
+        int maxResultCount,
+        BackgroundJobNameFilter? jobNameFilter)
+    {
+        return ObjectMapper.Map<List<BackgroundJobRecord>, List<BackgroundJobInfo>>(
+            await BackgroundJobRepository.GetWaitingListAsync(applicationName, maxResultCount, jobNameFilter)
+        );
+    }
+
     public virtual async Task DeleteAsync(Guid jobId)
     {
         await BackgroundJobRepository.DeleteAsync(jobId);
     }
 
+    public virtual async Task<int> DeleteAsync(
+        string? applicationName,
+        DateTime completedBefore,
+        int maxResultCount,
+        CancellationToken cancellationToken = default)
+    {
+        return await BackgroundJobRepository.DeleteAsync(applicationName, completedBefore, maxResultCount, cancellationToken);
+    }
+
     public virtual async Task UpdateAsync(BackgroundJobInfo jobInfo)
     {
         var backgroundJobRecord = await BackgroundJobRepository.FindAsync(jobInfo.Id);

modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs

@@ -5,7 +5,7 @@ using System.Threading.Tasks;
 using Riok.Mapperly.Abstractions;
 using Volo.Abp.Mapperly;
 
-namespace  Volo.Abp.BackgroundJobs;
+namespace Volo.Abp.BackgroundJobs;
 
 [Mapper(RequiredMappingStrategy = RequiredMappingStrategy.Target)]
 public partial class BackgroundJobInfoToBackgroundJobRecordMapper
@@ -31,6 +31,6 @@ public partial class BackgroundJobRecordToBackgroundJobInfoMapper
     : MapperBase<BackgroundJobRecord, BackgroundJobInfo>
 {
     public override partial BackgroundJobInfo Map(BackgroundJobRecord source);
-    
+
     public override partial void Map(BackgroundJobRecord source, BackgroundJobInfo destination);
 }

modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs

@@ -2,12 +2,26 @@
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
-using JetBrains.Annotations;
 using Volo.Abp.Domain.Repositories;
 
 namespace Volo.Abp.BackgroundJobs;
 
 public interface IBackgroundJobRepository : IBasicRepository<BackgroundJobRecord, Guid>
 {
-    Task<List<BackgroundJobRecord>> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default);
+    Task<List<BackgroundJobRecord>> GetWaitingListAsync(
+        string? applicationName,
+        int maxResultCount,
+        CancellationToken cancellationToken = default);
+
+    Task<List<BackgroundJobRecord>> GetWaitingListAsync(
+        string? applicationName,
+        int maxResultCount,
+        BackgroundJobNameFilter? jobNameFilter,
+        CancellationToken cancellationToken = default);
+
+    Task<int> DeleteAsync(
+        string? applicationName,
+        DateTime completedBefore,
+        int maxResultCount,
+        CancellationToken cancellationToken = default);
 }

modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs

@@ -29,9 +29,10 @@ public static class BackgroundJobsDbContextModelCreatingExtensions
             b.Property(x => x.NextTryTime);
             b.Property(x => x.LastTryTime);
             b.Property(x => x.IsAbandoned).HasDefaultValue(false);
+            b.Property(x => x.CompletionTime);
             b.Property(x => x.Priority).HasDefaultValue(BackgroundJobPriority.Normal).HasSentinel(BackgroundJobPriority.Normal);
 
-            b.HasIndex(x => new { x.IsAbandoned, x.NextTryTime });
+            b.HasIndex(x => new { x.ApplicationName, x.CompletionTime, x.IsAbandoned, x.NextTryTime });
 
             b.ApplyObjectExtensionMappings();
         });

modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs

@@ -3,7 +3,6 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using JetBrains.Annotations;
 using Microsoft.EntityFrameworkCore;
 using Volo.Abp.Domain.Repositories.EntityFrameworkCore;
 using Volo.Abp.EntityFrameworkCore;
@@ -23,20 +22,71 @@ public class EfCoreBackgroundJobRepository : EfCoreRepository<IBackgroundJobsDbC
         Clock = clock;
     }
 
-    public virtual async Task<List<BackgroundJobRecord>> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default)
+    public virtual Task<List<BackgroundJobRecord>> GetWaitingListAsync(
+        string? applicationName,
+        int maxResultCount,
+        CancellationToken cancellationToken = default)
     {
-        return await (await GetWaitingListQueryAsync(applicationName, maxResultCount)).ToListAsync(GetCancellationToken(cancellationToken));
+        return GetWaitingListAsync(applicationName, maxResultCount, null, cancellationToken);
     }
 
-    protected virtual async Task<IQueryable<BackgroundJobRecord>> GetWaitingListQueryAsync([CanBeNull] string applicationName, int maxResultCount)
+    public virtual async Task<List<BackgroundJobRecord>> GetWaitingListAsync(
+        string? applicationName,
+        int maxResultCount,
+        BackgroundJobNameFilter? jobNameFilter,
+        CancellationToken cancellationToken = default)
+    {
+        return await (await GetWaitingListQueryAsync(applicationName, maxResultCount, jobNameFilter))
+            .ToListAsync(GetCancellationToken(cancellationToken));
+    }
+
+    protected virtual async Task<IQueryable<BackgroundJobRecord>> GetWaitingListQueryAsync(
+        string? applicationName,
+        int maxResultCount,
+        BackgroundJobNameFilter? jobNameFilter = null)
     {
         var now = Clock.Now;
-        return (await GetDbSetAsync())
+        var filter = jobNameFilter ?? BackgroundJobNameFilter.None;
+        var jobNames = filter.JobNames.ToList();
+
+        var query = (await GetDbSetAsync())
             .Where(t => t.ApplicationName == applicationName)
-            .Where(t => !t.IsAbandoned && t.NextTryTime <= now)
+            .Where(t => !t.IsAbandoned && t.CompletionTime == null && t.NextTryTime <= now);
+
+        if (filter.Mode == BackgroundJobNameFilterMode.Include)
+        {
+            query = query.Where(t => jobNames.Contains(t.JobName));
+        }
+        else if (filter.Mode == BackgroundJobNameFilterMode.Exclude)
+        {
+            query = query.Where(t => !jobNames.Contains(t.JobName));
+        }
+
+        return query
             .OrderByDescending(t => t.Priority)
             .ThenBy(t => t.TryCount)
             .ThenBy(t => t.NextTryTime)
             .Take(maxResultCount);
     }
+
+    public virtual async Task<int> DeleteAsync(string? applicationName, DateTime completedBefore, int maxResultCount, CancellationToken cancellationToken = default)
+    {
+        var token = GetCancellationToken(cancellationToken);
+        var dbSet = await GetDbSetAsync();
+
+        var ids = await dbSet
+            .Where(t => t.ApplicationName == applicationName)
+            .Where(t => t.CompletionTime != null && t.CompletionTime < completedBefore)
+            .OrderBy(t => t.CompletionTime)
+            .Select(t => t.Id)
+            .Take(maxResultCount)
+            .ToListAsync(token);
+
+        if (ids.Count == 0)
+        {
+            return 0;
+        }
+
+        return await dbSet.Where(t => ids.Contains(t.Id)).ExecuteDeleteAsync(token);
+    }
 }

modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs

@@ -3,7 +3,6 @@ using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using System.Linq;
-using JetBrains.Annotations;
 using MongoDB.Driver;
 using MongoDB.Driver.Linq;
 using Volo.Abp.Domain.Repositories.MongoDB;
@@ -24,20 +23,78 @@ public class MongoBackgroundJobRepository : MongoDbRepository<IBackgroundJobsMon
         Clock = clock;
     }
 
-    public virtual async Task<List<BackgroundJobRecord>> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default)
+    public virtual Task<List<BackgroundJobRecord>> GetWaitingListAsync(
+        string? applicationName,
+        int maxResultCount,
+        CancellationToken cancellationToken = default)
     {
-        return await (await GetWaitingListQuery(applicationName, maxResultCount, cancellationToken)).ToListAsync(GetCancellationToken(cancellationToken));
+        return GetWaitingListAsync(applicationName, maxResultCount, null, cancellationToken);
     }
 
-    protected virtual async Task<IQueryable<BackgroundJobRecord>> GetWaitingListQuery([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default)
+    public virtual async Task<List<BackgroundJobRecord>> GetWaitingListAsync(
+        string? applicationName,
+        int maxResultCount,
+        BackgroundJobNameFilter? jobNameFilter,
+        CancellationToken cancellationToken = default)
+    {
+        return await (await GetWaitingListQuery(applicationName, maxResultCount, jobNameFilter, cancellationToken))
+            .ToListAsync(GetCancellationToken(cancellationToken));
+    }
+
+    protected virtual async Task<IQueryable<BackgroundJobRecord>> GetWaitingListQuery(
+        string? applicationName,
+        int maxResultCount,
+        BackgroundJobNameFilter? jobNameFilter = null,
+        CancellationToken cancellationToken = default)
     {
         var now = Clock.Now;
-        return (await GetQueryableAsync(cancellationToken))
+        var filter = jobNameFilter ?? BackgroundJobNameFilter.None;
+        var jobNames = filter.JobNames.ToList();
+
+        var query = (await GetQueryableAsync(cancellationToken))
             .Where(t => t.ApplicationName == applicationName)
-            .Where(t => !t.IsAbandoned && t.NextTryTime <= now)
+            .Where(t => !t.IsAbandoned && t.CompletionTime == null && t.NextTryTime <= now);
+
+        if (filter.Mode == BackgroundJobNameFilterMode.Include)
+        {
+            query = query.Where(t => jobNames.Contains(t.JobName));
+        }
+        else if (filter.Mode == BackgroundJobNameFilterMode.Exclude)
+        {
+            query = query.Where(t => !jobNames.Contains(t.JobName));
+        }
+
+        return query
             .OrderByDescending(t => t.Priority)
             .ThenBy(t => t.TryCount)
             .ThenBy(t => t.NextTryTime)
             .Take(maxResultCount);
     }
+
+    public virtual async Task<int> DeleteAsync(string? applicationName, DateTime completedBefore, int maxResultCount, CancellationToken cancellationToken = default)
+    {
+        var token = GetCancellationToken(cancellationToken);
+
+        var ids = await (await GetQueryableAsync(token))
+            .Where(t => t.ApplicationName == applicationName)
+            .Where(t => t.CompletionTime != null && t.CompletionTime < completedBefore)
+            .OrderBy(t => t.CompletionTime)
+            .Select(t => t.Id)
+            .Take(maxResultCount)
+            .ToListAsync(token);
+
+        if (ids.Count == 0)
+        {
+            return 0;
+        }
+
+        var dbContext = await GetDbContextAsync(token);
+        var collection = dbContext.Collection<BackgroundJobRecord>();
+
+        var result = dbContext.SessionHandle != null
+            ? await collection.DeleteManyAsync(dbContext.SessionHandle, x => ids.Contains(x.Id), cancellationToken: token)
+            : await collection.DeleteManyAsync(x => ids.Contains(x.Id), cancellationToken: token);
+
+        return (int)result.DeletedCount;
+    }
 }