Add the database-backed job runner

Directory.Packages.props

@@ -5,6 +5,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageVersion Include="Cronos" Version="0.13.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.PublicApiAnalyzers" Version="5.6.0" />
     <PackageVersion Include="Microsoft.EntityFrameworkCore" Version="10.0.14" />
     <PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.14" />

src/FluentCMS.Core/DependencyInjection/FluentCmsBuilder.cs

@@ -1,5 +1,6 @@
 using System.Reflection;
 using FluentCMS.Core.Authorization;
+using FluentCMS.Core.Jobs;
 using FluentCMS.Core.Storage;
 using Microsoft.Extensions.DependencyInjection;
 
@@ -10,12 +11,15 @@ public sealed class FluentCmsBuilder
     internal FluentCmsBuilder(IServiceCollection services)
     {
         Services = services;
+        Jobs = new JobsBuilder(services);
     }
 
     public IServiceCollection Services { get; }
 
     public AuthorizationBuilder Authorization { get; } = new();
 
+    public JobsBuilder Jobs { get; }
+
     internal IDataProvider? DataProvider { get; private set; }
 
     internal Assembly? ContentMigrationsAssembly { get; private set; }

src/FluentCMS.Core/DependencyInjection/FluentCmsServiceCollectionExtensions.cs

@@ -4,10 +4,12 @@
 using FluentCMS.Core.Authorization;
 using FluentCMS.Core.Content;
 using FluentCMS.Core.Content.Query;
+using FluentCMS.Core.Jobs;
 using FluentCMS.Core.Localization;
 using FluentCMS.Core.Schema;
 using FluentCMS.Core.Sites;
 using FluentCMS.Core.Storage;
+using FluentCMS.Jobs;
 using FluentCMS.Localization;
 using FluentCMS.Schema;
 using FluentCMS.Sites;
@@ -81,6 +83,17 @@ public static IServiceCollection AddFluentCms(this IServiceCollection services,
         services.AddScoped<LocaleFallbackReader>();
         services.AddScoped<SiteManager>();
 
+        services.AddSingleton(builder.Jobs.Options);
+        services.AddSingleton<JobRegistry>();
+        services.AddSingleton<JobStore>();
+        services.AddSingleton<JobExecutor>();
+        services.AddSingleton<RecurringJobs>();
+        services.TryAddSingleton<IJobEventSink, NullJobEventSink>();
+        services.TryAddSingleton<IJobScheduler, JobScheduler>();
+        services.TryAddSingleton<IJobRunner, JobRunner>();
+        services.TryAddSingleton<IJobDeadLetters, JobDeadLetters>();
+        services.AddHostedService<JobRunnerHostedService>();
+
         return services;
     }
 }

src/FluentCMS.Core/FluentCMS.Core.csproj

@@ -6,6 +6,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <PackageReference Include="Cronos" />
     <PackageReference Include="Microsoft.EntityFrameworkCore" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.Relational" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" />

src/FluentCMS.Core/Jobs/JobDeadLetters.cs

@@ -0,0 +1,49 @@
+using FluentCMS.Core.Storage;
+using FluentCMS.Core.Storage.Entities;
+using FluentCMS.Jobs;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace FluentCMS.Core.Jobs;
+
+internal sealed class JobDeadLetters(IServiceScopeFactory scopes, JobRunnerOptions options, TimeProvider time) : IJobDeadLetters
+{
+    public async Task<IReadOnlyList<DeadLetteredJob>> ListAsync(int offset = 0, int limit = 50, CancellationToken cancellationToken = default)
+    {
+        ArgumentOutOfRangeException.ThrowIfNegative(offset);
+        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(limit);
+
+        await using var scope = scopes.CreateAsyncScope();
+        var context = scope.ServiceProvider.GetRequiredService<SystemDbContext>();
+
+        return await context.Set<Job>().AsNoTracking()
+            .Where(job => job.Status == JobStatus.DeadLettered)
+            .OrderByDescending(job => job.CompletedAt)
+            .ThenBy(job => job.Id)
+            .Skip(offset)
+            .Take(limit)
+            .Select(job => new DeadLetteredJob(job.Id, job.Type, job.Queue, job.Attempts, job.LastError, job.CompletedAt))
+            .ToListAsync(cancellationToken);
+    }
+
+    // Attempts count on rather than start over, because cms_job_attempt keeps the rows of the earlier ones.
+    public async Task<bool> RetryAsync(Guid id, CancellationToken cancellationToken = default)
+    {
+        await using var scope = scopes.CreateAsyncScope();
+        var context = scope.ServiceProvider.GetRequiredService<SystemDbContext>();
+
+        var now = time.GetUtcNow();
+        var further = options.DefaultMaxAttempts;
+        var queued = await context.Set<Job>()
+            .Where(job => job.Id == id && job.Status == JobStatus.DeadLettered)
+            .ExecuteUpdateAsync(
+                job => job
+                    .SetProperty(row => row.Status, JobStatus.Pending)
+                    .SetProperty(row => row.RunAt, now)
+                    .SetProperty(row => row.MaxAttempts, row => row.Attempts + further)
+                    .SetProperty(row => row.CompletedAt, (DateTimeOffset?)null),
+                cancellationToken);
+
+        return queued == 1;
+    }
+}

src/FluentCMS.Core/Jobs/JobErrorCodes.cs

@@ -0,0 +1,12 @@
+namespace FluentCMS.Core.Jobs;
+
+internal static class JobErrorCodes
+{
+    public const string UnknownJobType = "FCMS2501";
+    public const string BadPayload = "FCMS2502";
+    public const string InvalidCron = "FCMS2503";
+    public const string UnregisteredJob = "FCMS2504";
+    public const string DuplicateJobType = "FCMS2505";
+    public const string InvalidRunnerOptions = "FCMS2506";
+    public const string BadActor = "FCMS2507";
+}

src/FluentCMS.Core/Jobs/JobErrorText.cs

@@ -0,0 +1,109 @@
+using System.Text;
+using System.Text.Json;
+using System.Text.RegularExpressions;
+using FluentCMS.Core.Storage.Entities;
+
+namespace FluentCMS.Core.Jobs;
+
+// An exception message can quote the job's input, so every string, number and property name of the payload is taken
+// out of what is stored and logged. This cannot catch a secret the job has encoded or derived.
+internal static class JobErrorText
+{
+    private const string Redacted = "***";
+    private const string Truncated = "...";
+
+    // A shorter value is matched only as a whole word, since inside a word it is more likely chance than a quote.
+    private const int AnywhereFromLength = 4;
+
+    public static string ForStorage(Exception exception, string payload)
+    {
+        if (exception is JobInvocationException)
+        {
+            return Cap(exception.Message);
+        }
+
+        var values = ValuesOf(payload);
+        var text = new StringBuilder();
+        for (var current = exception; current is not null; current = current.InnerException)
+        {
+            text.Append(text.Length == 0 ? "" : " ---> ").Append(current.GetType().FullName).Append(": ").Append(Scrub(current.Message, values));
+        }
+
+        return Cap(text.ToString());
+    }
+
+    public static string ForLog(Exception exception, string payload) => Scrub(exception.ToString(), ValuesOf(payload));
+
+    public static string Cap(string text)
+    {
+        if (text.Length <= Job.LastErrorMaxLength)
+        {
+            return text;
+        }
+
+        var keep = Job.LastErrorMaxLength - Truncated.Length;
+        if (char.IsHighSurrogate(text[keep - 1]))
+        {
+            keep--;
+        }
+
+        return string.Concat(text.AsSpan(0, keep), Truncated);
+    }
+
+    private static string Scrub(string text, List<string> values)
+    {
+        foreach (var value in values)
+        {
+            text = value.Length >= AnywhereFromLength
+                ? text.Replace(value, Redacted, StringComparison.Ordinal)
+                : Regex.Replace(text, $@"(?<![\p{{L}}\p{{N}}]){Regex.Escape(value)}(?![\p{{L}}\p{{N}}])", Redacted, RegexOptions.CultureInvariant);
+        }
+
+        return text;
+    }
+
+    // The longest first, so that a value inside another does not leave the rest of it behind.
+    private static List<string> ValuesOf(string payload)
+    {
+        var values = new List<string>();
+        try
+        {
+            using var document = JsonDocument.Parse(payload);
+            Collect(document.RootElement, values);
+        }
+        catch (JsonException)
+        {
+            values.Add(payload);
+        }
+
+        return [.. values.Where(value => value.Length > 0).Distinct(StringComparer.Ordinal).OrderByDescending(value => value.Length)];
+    }
+
+    private static void Collect(JsonElement element, List<string> values)
+    {
+        switch (element.ValueKind)
+        {
+            case JsonValueKind.String:
+                values.Add(element.GetString()!);
+                break;
+            case JsonValueKind.Number:
+                values.Add(element.GetRawText());
+                break;
+            case JsonValueKind.Array:
+                foreach (var item in element.EnumerateArray())
+                {
+                    Collect(item, values);
+                }
+
+                break;
+            case JsonValueKind.Object:
+                foreach (var property in element.EnumerateObject())
+                {
+                    values.Add(property.Name);
+                    Collect(property.Value, values);
+                }
+
+                break;
+        }
+    }
+}

src/FluentCMS.Core/Jobs/JobEvents.cs

@@ -0,0 +1,21 @@
+namespace FluentCMS.Core.Jobs;
+
+// The events of spec 4.11, until the event bus (#2697) publishes them.
+internal sealed record JobFailedEvent(Guid JobId, string Type, string Queue, int Attempt, int MaxAttempts, string Error);
+
+internal sealed record JobDeadLetteredEvent(Guid JobId, string Type, string Queue, int Attempts, string Error);
+
+// The seam for #2697. A sink that throws is logged and changes nothing about the job.
+internal interface IJobEventSink
+{
+    Task PublishAsync(JobFailedEvent failed);
+
+    Task PublishAsync(JobDeadLetteredEvent deadLettered);
+}
+
+internal sealed class NullJobEventSink : IJobEventSink
+{
+    public Task PublishAsync(JobFailedEvent failed) => Task.CompletedTask;
+
+    public Task PublishAsync(JobDeadLetteredEvent deadLettered) => Task.CompletedTask;
+}