diff --git a/Directory.Packages.props b/Directory.Packages.props
index 041b190c6..fa8e6acf9 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -5,6 +5,7 @@
+
diff --git a/src/FluentCMS.Core/DependencyInjection/FluentCmsBuilder.cs b/src/FluentCMS.Core/DependencyInjection/FluentCmsBuilder.cs
index a3562fac6..aac783f6c 100644
--- a/src/FluentCMS.Core/DependencyInjection/FluentCmsBuilder.cs
+++ b/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; }
diff --git a/src/FluentCMS.Core/DependencyInjection/FluentCmsServiceCollectionExtensions.cs b/src/FluentCMS.Core/DependencyInjection/FluentCmsServiceCollectionExtensions.cs
index a6a993f0c..a44f6923c 100644
--- a/src/FluentCMS.Core/DependencyInjection/FluentCmsServiceCollectionExtensions.cs
+++ b/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();
services.AddScoped();
+ services.AddSingleton(builder.Jobs.Options);
+ services.AddSingleton();
+ services.AddSingleton();
+ services.AddSingleton();
+ services.AddSingleton();
+ services.TryAddSingleton();
+ services.TryAddSingleton();
+ services.TryAddSingleton();
+ services.TryAddSingleton();
+ services.AddHostedService();
+
return services;
}
}
diff --git a/src/FluentCMS.Core/FluentCMS.Core.csproj b/src/FluentCMS.Core/FluentCMS.Core.csproj
index 059b2f6e1..fa3823f3e 100644
--- a/src/FluentCMS.Core/FluentCMS.Core.csproj
+++ b/src/FluentCMS.Core/FluentCMS.Core.csproj
@@ -6,6 +6,7 @@
+
diff --git a/src/FluentCMS.Core/Jobs/JobDeadLetters.cs b/src/FluentCMS.Core/Jobs/JobDeadLetters.cs
new file mode 100644
index 000000000..a08ed8afe
--- /dev/null
+++ b/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> 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();
+
+ return await context.Set().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 RetryAsync(Guid id, CancellationToken cancellationToken = default)
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+
+ var now = time.GetUtcNow();
+ var further = options.DefaultMaxAttempts;
+ var queued = await context.Set()
+ .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;
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/JobErrorCodes.cs b/src/FluentCMS.Core/Jobs/JobErrorCodes.cs
new file mode 100644
index 000000000..bfa7ae65b
--- /dev/null
+++ b/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";
+}
diff --git a/src/FluentCMS.Core/Jobs/JobErrorText.cs b/src/FluentCMS.Core/Jobs/JobErrorText.cs
new file mode 100644
index 000000000..ec42898f6
--- /dev/null
+++ b/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 values)
+ {
+ foreach (var value in values)
+ {
+ text = value.Length >= AnywhereFromLength
+ ? text.Replace(value, Redacted, StringComparison.Ordinal)
+ : Regex.Replace(text, $@"(? ValuesOf(string payload)
+ {
+ var values = new List();
+ 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 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;
+ }
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/JobEvents.cs b/src/FluentCMS.Core/Jobs/JobEvents.cs
new file mode 100644
index 000000000..9f26ddd54
--- /dev/null
+++ b/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;
+}
diff --git a/src/FluentCMS.Core/Jobs/JobExecutor.cs b/src/FluentCMS.Core/Jobs/JobExecutor.cs
new file mode 100644
index 000000000..fca9d91e1
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobExecutor.cs
@@ -0,0 +1,228 @@
+using FluentCMS.Content;
+using FluentCMS.Core.Sites;
+using FluentCMS.Core.Storage;
+using FluentCMS.Core.Storage.Entities;
+using FluentCMS.Jobs;
+using FluentCMS.Sites;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+namespace FluentCMS.Core.Jobs;
+
+internal sealed class JobExecutor(
+ JobStore store,
+ JobRegistry registry,
+ IServiceScopeFactory scopes,
+ IOperationContextAccessor operation,
+ JobRunnerOptions options,
+ TimeProvider time,
+ ILogger logger)
+{
+ private static readonly TimeSpan FirstSettleRetry = TimeSpan.FromSeconds(1);
+ private static readonly TimeSpan LastSettleRetry = TimeSpan.FromSeconds(30);
+
+ // False when the queue has nothing due.
+ public async Task RunNextAsync(
+ string queue,
+ string owner,
+ CancellationToken stopClaiming = default,
+ CancellationToken shutdown = default,
+ CancellationToken abandon = default)
+ {
+ if (await store.ClaimAsync(queue, owner, stopClaiming) is not { } job)
+ {
+ return false;
+ }
+
+ await RunAsync(job, shutdown, abandon);
+ return true;
+ }
+
+ public async Task RunAsync(ClaimedJob job, CancellationToken shutdown = default, CancellationToken abandon = default)
+ {
+ using var attempt = CancellationTokenSource.CreateLinkedTokenSource(shutdown);
+ using var stopHeartbeat = CancellationTokenSource.CreateLinkedTokenSource(abandon);
+
+ var heartbeat = HeartbeatAsync(job, attempt, stopHeartbeat.Token);
+ try
+ {
+ await ExecuteAndSettleAsync(job, shutdown, attempt.Token, abandon);
+ }
+ finally
+ {
+ await stopHeartbeat.CancelAsync();
+ await heartbeat;
+ }
+ }
+
+ private async Task ExecuteAndSettleAsync(ClaimedJob job, CancellationToken shutdown, CancellationToken attempt, CancellationToken abandon)
+ {
+ Func> settle;
+ try
+ {
+ await InvokeAsync(job, attempt);
+ settle = () => store.SucceedAsync(job);
+ }
+ catch (OperationCanceledException) when (shutdown.IsCancellationRequested)
+ {
+ JobLog.Released(logger, job.Id, job.Type, job.Attempt);
+ settle = () => store.ReleaseAsync(job);
+ }
+ catch (OperationCanceledException) when (attempt.IsCancellationRequested)
+ {
+ JobLog.LeaseLost(logger, job.Id, job.Type, job.Attempt);
+ return;
+ }
+ catch (Exception exception)
+ {
+ JobLog.AttemptFailed(logger, job.Id, job.Type, job.Attempt, JobErrorText.ForLog(exception, job.Payload));
+ var error = JobErrorText.ForStorage(exception, job.Payload);
+ settle = () => store.FailAsync(job, error);
+ }
+
+ await SettleWithRetriesAsync(job, settle, abandon);
+ }
+
+ // A result not written means the job runs again, or is dead-lettered as if it never finished, so a brief outage is
+ // waited out. Retrying past the real end of the lease is safe, because a write matches nothing once another worker
+ // has the job. It gives up after one lease, in case the failure is not an outage.
+ private async Task SettleWithRetriesAsync(ClaimedJob job, Func> settle, CancellationToken abandon)
+ {
+ var giveUpAt = time.GetUtcNow() + options.LeaseDuration;
+ var delay = FirstSettleRetry;
+
+ while (true)
+ {
+ try
+ {
+ if (!await settle())
+ {
+ JobLog.LeaseLost(logger, job.Id, job.Type, job.Attempt);
+ }
+
+ return;
+ }
+ catch (Exception exception) when (time.GetUtcNow() + delay < giveUpAt)
+ {
+ JobLog.SettleFailed(logger, exception, job.Id);
+ }
+
+ await Task.Delay(delay, time, abandon);
+ delay = TimeSpan.FromTicks(Math.Min(delay.Ticks * 2, LastSettleRetry.Ticks));
+ }
+ }
+
+ private async Task InvokeAsync(ClaimedJob job, CancellationToken attempt)
+ {
+ var registration = registry.Find(job.Type)
+ ?? throw new JobInvocationException($"{JobErrorCodes.UnknownJobType}: no job is registered for the type '{job.Type}'.");
+
+ var payload = ReadPayload(registration, job);
+
+ await using var scope = scopes.CreateAsyncScope();
+ var context = new JobContext(job.Id, job.Type, job.Queue, job.Attempt, job.MaxAttempts, await OperationOfAsync(job, payload, scope.ServiceProvider), attempt);
+
+ var previous = operation.Current;
+ operation.Current = context.Operation;
+ try
+ {
+ await registration.Invoke(scope.ServiceProvider, payload, context);
+ }
+ finally
+ {
+ operation.Current = previous;
+ }
+ }
+
+ // Keeps nothing of the reader's exception: a converter can quote any value in it, and a path can name a dictionary key.
+ private static object ReadPayload(JobRegistration registration, ClaimedJob job)
+ {
+ try
+ {
+ return registration.Read(job.Payload)
+ ?? throw new JobInvocationException($"{JobErrorCodes.BadPayload}: the payload of '{job.Type}' is null.");
+ }
+ catch (Exception exception) when (exception is not JobInvocationException)
+ {
+ throw new JobInvocationException(
+ $"{JobErrorCodes.BadPayload}: the payload of '{job.Type}' could not be read as {registration.Payload.Name} ({exception.GetType().Name}).");
+ }
+ }
+
+ private async Task OperationOfAsync(ClaimedJob job, object payload, IServiceProvider services)
+ {
+ var site = job.SiteId ?? DefaultSite.Id;
+ var locale = await services.GetRequiredService().Set()
+ .Where(row => row.Id == site)
+ .Select(row => row.DefaultLocale)
+ .FirstOrDefaultAsync();
+
+ return new OperationContext(Principal.System, Origin.Job, new SiteId(site), locale ?? DefaultSite.Locale, ClientId: null, job.Id.ToString(), time.GetUtcNow())
+ {
+ Actor = ActorOf(job, payload),
+ };
+ }
+
+ // Built from the id alone, so that nothing in a payload can bring claims. It never replaces Principal.System.
+ private static Principal? ActorOf(ClaimedJob job, object payload)
+ {
+ if (payload is not IJobActorPayload { Actor: { } actor })
+ {
+ return null;
+ }
+
+ if (actor.Id == Guid.Empty)
+ {
+ throw new JobInvocationException($"{JobErrorCodes.BadActor}: the payload of '{job.Type}' names an actor with an empty id.");
+ }
+
+ try
+ {
+ return new Principal(actor.Kind, actor.Id, actor.Name);
+ }
+ catch (ArgumentException)
+ {
+ throw new JobInvocationException($"{JobErrorCodes.BadActor}: the payload of '{job.Type}' names an actor of kind {actor.Kind}, which is not an identity a job can act for.");
+ }
+ }
+
+ private async Task HeartbeatAsync(ClaimedJob job, CancellationTokenSource attempt, CancellationToken stop)
+ {
+ using var timer = new PeriodicTimer(options.HeartbeatInterval, time);
+ var leaseUntil = job.LeaseUntil;
+ try
+ {
+ while (await timer.WaitForNextTickAsync(stop))
+ {
+ try
+ {
+ if (await store.RenewLeaseAsync(job) is not { } renewed)
+ {
+ await attempt.CancelAsync();
+ return;
+ }
+
+ leaseUntil = renewed;
+ }
+ catch (Exception exception) when (!stop.IsCancellationRequested)
+ {
+ JobLog.HeartbeatFailed(logger, exception, job.Id);
+
+ // With no renewal before the lease ends, another worker can take the job, so stop first.
+ if (time.GetUtcNow() + options.HeartbeatInterval >= leaseUntil)
+ {
+ await attempt.CancelAsync();
+ return;
+ }
+ }
+ }
+ }
+ catch (OperationCanceledException) when (stop.IsCancellationRequested)
+ {
+ }
+ }
+}
+
+// The job could not start. The message is what last_error stores, and never holds a value of the payload.
+internal sealed class JobInvocationException(string message) : Exception(message);
diff --git a/src/FluentCMS.Core/Jobs/JobLog.cs b/src/FluentCMS.Core/Jobs/JobLog.cs
new file mode 100644
index 000000000..78bd6856b
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobLog.cs
@@ -0,0 +1,43 @@
+using Microsoft.Extensions.Logging;
+
+namespace FluentCMS.Core.Jobs;
+
+// What a job threw arrives as text already scrubbed of the payload, never as the exception itself.
+internal static partial class JobLog
+{
+ [LoggerMessage(Level = LogLevel.Warning, Message = "Job {JobId} ({Type}) failed attempt {Attempt}: {Error}")]
+ public static partial void AttemptFailed(ILogger logger, Guid jobId, string type, int attempt, string error);
+
+ [LoggerMessage(Level = LogLevel.Error, Message = "Job {JobId} ({Type}) was dead-lettered after {Attempts} attempts: {Error}")]
+ public static partial void DeadLettered(ILogger logger, Guid jobId, string type, int attempts, string error);
+
+ [LoggerMessage(Level = LogLevel.Warning, Message = "Job {JobId} ({Type}) lost its lease during attempt {Attempt}, so this worker leaves it alone.")]
+ public static partial void LeaseLost(ILogger logger, Guid jobId, string type, int attempt);
+
+ [LoggerMessage(Level = LogLevel.Information, Message = "Job {JobId} ({Type}) was released by the shutdown during attempt {Attempt}.")]
+ public static partial void Released(ILogger logger, Guid jobId, string type, int attempt);
+
+ [LoggerMessage(Level = LogLevel.Warning, Message = "Could not renew the lease of job {JobId}.")]
+ public static partial void HeartbeatFailed(ILogger logger, Exception exception, Guid jobId);
+
+ [LoggerMessage(Level = LogLevel.Warning, Message = "Could not write the outcome of job {JobId}; trying again.")]
+ public static partial void SettleFailed(ILogger logger, Exception exception, Guid jobId);
+
+ [LoggerMessage(Level = LogLevel.Error, Message = "The worker of queue {Queue} hit an error and carries on.")]
+ public static partial void WorkerFailed(ILogger logger, Exception exception, string queue);
+
+ [LoggerMessage(Level = LogLevel.Error, Message = "The maintenance of the job runner hit an error and carries on.")]
+ public static partial void MaintenanceFailed(ILogger logger, Exception exception);
+
+ [LoggerMessage(Level = LogLevel.Error, Message = "A job event sink threw for job {JobId}.")]
+ public static partial void EventSinkFailed(ILogger logger, Exception exception, Guid jobId);
+
+ [LoggerMessage(Level = LogLevel.Error, Message = "Recurring job {Name} has a cron expression that cannot fire, so it is skipped: {Reason}")]
+ public static partial void RecurringCronInvalid(ILogger logger, string name, string reason);
+
+ [LoggerMessage(Level = LogLevel.Error, Message = "Recurring job {Name} could not fire; it tries again next time.")]
+ public static partial void RecurringFireFailed(ILogger logger, Exception exception, string name);
+
+ [LoggerMessage(Level = LogLevel.Warning, Message = "The job runner stopped with jobs still running, whose leases will expire.")]
+ public static partial void StoppedWithRunningJobs(ILogger logger);
+}
diff --git a/src/FluentCMS.Core/Jobs/JobRegistry.cs b/src/FluentCMS.Core/Jobs/JobRegistry.cs
new file mode 100644
index 000000000..b4b7472c6
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobRegistry.cs
@@ -0,0 +1,73 @@
+using System.Collections.Frozen;
+using System.Text.Json;
+using FluentCMS.Core.DependencyInjection;
+using FluentCMS.Jobs;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace FluentCMS.Core.Jobs;
+
+internal sealed class JobRegistration
+{
+ private JobRegistration(string type, Type job, Type payload, Func read, Func invoke)
+ {
+ Type = type;
+ Job = job;
+ Payload = payload;
+ Read = read;
+ Invoke = invoke;
+ }
+
+ public string Type { get; }
+
+ public Type Job { get; }
+
+ public Type Payload { get; }
+
+ public Func Read { get; }
+
+ public Func Invoke { get; }
+
+ public static JobRegistration Create(string type)
+ where TJob : class, IJob =>
+ new(
+ type,
+ typeof(TJob),
+ typeof(TPayload),
+ json => JsonSerializer.Deserialize(json, JobJson.Options),
+ (services, payload, context) => services.GetRequiredService().ExecuteAsync((TPayload)payload, context));
+}
+
+internal static class JobJson
+{
+ public static readonly JsonSerializerOptions Options = JsonSerializerOptions.Web;
+}
+
+internal sealed class JobRegistry
+{
+ private readonly FrozenDictionary _byType;
+ private readonly FrozenDictionary<(Type Job, Type Payload), string> _byJob;
+
+ public JobRegistry(IEnumerable registrations)
+ {
+ var all = registrations.ToList();
+ var duplicate = all.GroupBy(registration => registration.Type, StringComparer.Ordinal).FirstOrDefault(group => group.Count() > 1);
+ if (duplicate is not null)
+ {
+ throw new FluentCmsConfigurationException(JobErrorCodes.DuplicateJobType, $"The job type '{duplicate.Key}' is registered {duplicate.Count()} times.");
+ }
+
+ _byType = all.ToFrozenDictionary(registration => registration.Type, StringComparer.Ordinal);
+ _byJob = all
+ .GroupBy(registration => (registration.Job, registration.Payload))
+ .ToFrozenDictionary(group => group.Key, group => group.First().Type);
+ }
+
+ public JobRegistration? Find(string type) => _byType.GetValueOrDefault(type);
+
+ public string TypeOf() =>
+ _byJob.TryGetValue((typeof(TJob), typeof(TPayload)), out var type)
+ ? type
+ : throw new FluentCmsConfigurationException(
+ JobErrorCodes.UnregisteredJob,
+ $"{typeof(TJob).FullName} is not registered for the payload {typeof(TPayload).FullName}; call cms.Jobs.AddJob inside AddFluentCms.");
+}
diff --git a/src/FluentCMS.Core/Jobs/JobRunner.cs b/src/FluentCMS.Core/Jobs/JobRunner.cs
new file mode 100644
index 000000000..05ddde28c
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobRunner.cs
@@ -0,0 +1,176 @@
+using FluentCMS.Jobs;
+using Microsoft.Extensions.Logging;
+
+namespace FluentCMS.Core.Jobs;
+
+internal sealed class JobRunner(
+ JobExecutor executor,
+ JobStore store,
+ RecurringJobs recurring,
+ JobRunnerOptions options,
+ TimeProvider time,
+ ILogger logger) : IJobRunner, IDisposable
+{
+ private static readonly TimeSpan DisposeWait = TimeSpan.FromSeconds(5);
+
+ private readonly CancellationTokenSource _stopping = new();
+
+ private readonly CancellationTokenSource _shutdown = new();
+
+ // Stops the heartbeats of jobs that ignored the shutdown, so that their leases can expire.
+ private readonly CancellationTokenSource _abandon = new();
+
+ private readonly string _instance = Guid.NewGuid().ToString("N");
+
+ private Task? _loops;
+
+ public Task StartAsync(CancellationToken cancellationToken = default)
+ {
+ options.Validate();
+
+ if (_loops is not null)
+ {
+ throw new InvalidOperationException("The job runner has already started.");
+ }
+
+ var loops = new List { Task.Run(MaintainAsync, CancellationToken.None) };
+ foreach (var (queue, workers) in options.Queues)
+ {
+ for (var worker = 0; worker < workers; worker++)
+ {
+ var owner = $"{_instance}/{queue}/{worker}";
+ loops.Add(Task.Run(() => WorkAsync(queue, owner), CancellationToken.None));
+ }
+ }
+
+ _loops = Task.WhenAll(loops);
+ return Task.CompletedTask;
+ }
+
+ public async Task StopAsync(CancellationToken cancellationToken = default)
+ {
+ if (_loops is null)
+ {
+ return;
+ }
+
+ await _stopping.CancelAsync();
+
+ if (await FinishedWithinAsync(options.ShutdownGracePeriod, cancellationToken))
+ {
+ return;
+ }
+
+ await _shutdown.CancelAsync();
+
+ // The host may have stopped waiting, but a job that stops now still gets its release written.
+ if (!await FinishedWithinAsync(options.ReleaseTimeout, CancellationToken.None))
+ {
+ JobLog.StoppedWithRunningJobs(logger);
+ await _abandon.CancelAsync();
+ }
+ }
+
+ // A host disposed without being stopped must not leave a loop that still opens the database.
+ public void Dispose()
+ {
+ _stopping.Cancel();
+ _shutdown.Cancel();
+ _abandon.Cancel();
+
+ try
+ {
+ if (_loops?.Wait(DisposeWait) == false)
+ {
+ return;
+ }
+ }
+ catch (AggregateException)
+ {
+ }
+
+ _stopping.Dispose();
+ _shutdown.Dispose();
+ _abandon.Dispose();
+ }
+
+ private async Task FinishedWithinAsync(TimeSpan period, CancellationToken cancellationToken)
+ {
+ try
+ {
+ await _loops!.WaitAsync(period, time, cancellationToken);
+ return true;
+ }
+ catch (Exception exception) when (exception is TimeoutException or OperationCanceledException)
+ {
+ return false;
+ }
+ }
+
+ private async Task WorkAsync(string queue, string owner)
+ {
+ while (!_stopping.IsCancellationRequested)
+ {
+ var worked = false;
+ try
+ {
+ worked = await executor.RunNextAsync(queue, owner, _stopping.Token, _shutdown.Token, _abandon.Token);
+ }
+ catch (OperationCanceledException) when (_stopping.IsCancellationRequested)
+ {
+ return;
+ }
+ catch (Exception exception)
+ {
+ JobLog.WorkerFailed(logger, exception, queue);
+ }
+
+ if (!worked && !await WaitAsync(options.PollInterval))
+ {
+ return;
+ }
+ }
+ }
+
+ private async Task MaintainAsync()
+ {
+ while (!_stopping.IsCancellationRequested)
+ {
+ await MaintainAsync(() => recurring.FireDueAsync(_stopping.Token));
+ await MaintainAsync(store.SweepAsync);
+
+ if (!await WaitAsync(options.MaintenanceInterval))
+ {
+ return;
+ }
+ }
+ }
+
+ private async Task MaintainAsync(Func step)
+ {
+ try
+ {
+ await step();
+ }
+ catch (OperationCanceledException) when (_stopping.IsCancellationRequested)
+ {
+ }
+ catch (Exception exception)
+ {
+ JobLog.MaintenanceFailed(logger, exception);
+ }
+ }
+
+ private async Task WaitAsync(TimeSpan period)
+ {
+ try
+ {
+ await Task.Delay(period, time, _stopping.Token);
+ return true;
+ }
+ catch (OperationCanceledException)
+ {
+ return false;
+ }
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/JobRunnerHostedService.cs b/src/FluentCMS.Core/Jobs/JobRunnerHostedService.cs
new file mode 100644
index 000000000..5ec3497de
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobRunnerHostedService.cs
@@ -0,0 +1,11 @@
+using FluentCMS.Jobs;
+using Microsoft.Extensions.Hosting;
+
+namespace FluentCMS.Core.Jobs;
+
+internal sealed class JobRunnerHostedService(IJobRunner runner) : IHostedService
+{
+ public Task StartAsync(CancellationToken cancellationToken) => runner.StartAsync(cancellationToken);
+
+ public Task StopAsync(CancellationToken cancellationToken) => runner.StopAsync(cancellationToken);
+}
diff --git a/src/FluentCMS.Core/Jobs/JobRunnerOptions.cs b/src/FluentCMS.Core/Jobs/JobRunnerOptions.cs
new file mode 100644
index 000000000..058ec8e48
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobRunnerOptions.cs
@@ -0,0 +1,65 @@
+using FluentCMS.Core.DependencyInjection;
+
+namespace FluentCMS.Core.Jobs;
+
+public sealed class JobRunnerOptions
+{
+ public const string DefaultQueue = "default";
+
+ private static readonly TimeSpan LongestWait = TimeSpan.FromDays(49);
+
+ // Workers per queue on this instance, which is also the concurrency limit of the queue here. A queue left out has
+ // no worker on this instance.
+ public IDictionary Queues { get; } = new Dictionary(StringComparer.Ordinal) { [DefaultQueue] = 2 };
+
+ public int DefaultMaxAttempts { get; set; } = 5;
+
+ public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(5);
+
+ public TimeSpan LeaseDuration { get; set; } = TimeSpan.FromMinutes(5);
+
+ public TimeSpan HeartbeatInterval { get; set; } = TimeSpan.FromMinutes(1);
+
+ public TimeSpan MaintenanceInterval { get; set; } = TimeSpan.FromSeconds(15);
+
+ public TimeSpan RetryBaseDelay { get; set; } = TimeSpan.FromSeconds(30);
+
+ public TimeSpan RetryMaxDelay { get; set; } = TimeSpan.FromHours(1);
+
+ // A retry waits up to this fraction longer, so that jobs failing together do not return together.
+ public double RetryJitter { get; set; } = 0.2;
+
+ public TimeSpan ShutdownGracePeriod { get; set; } = TimeSpan.FromSeconds(20);
+
+ public TimeSpan ReleaseTimeout { get; set; } = TimeSpan.FromSeconds(5);
+
+ internal void Validate()
+ {
+ foreach (var (queue, workers) in Queues)
+ {
+ Require(!string.IsNullOrEmpty(queue) && queue.Length <= JobScheduler.QueueMaxLength, $"The queue name '{queue}' is not valid.");
+ Require(workers >= 0, $"Queue '{queue}' has {workers} workers.");
+ }
+
+ Require(DefaultMaxAttempts >= 1, $"{nameof(DefaultMaxAttempts)} is {DefaultMaxAttempts}; it must be at least 1.");
+ Require(PollInterval > TimeSpan.Zero, $"{nameof(PollInterval)} must be positive.");
+ Require(MaintenanceInterval > TimeSpan.Zero, $"{nameof(MaintenanceInterval)} must be positive.");
+ Require(HeartbeatInterval > TimeSpan.Zero, $"{nameof(HeartbeatInterval)} must be positive.");
+ Require(LeaseDuration > HeartbeatInterval, $"{nameof(LeaseDuration)} must be longer than {nameof(HeartbeatInterval)}, or a live job loses its lease.");
+ Require(RetryBaseDelay >= TimeSpan.Zero && RetryMaxDelay >= RetryBaseDelay, $"{nameof(RetryMaxDelay)} must be at least {nameof(RetryBaseDelay)}, which must not be negative.");
+ Require(RetryJitter >= 0, $"{nameof(RetryJitter)} must not be negative.");
+ Require(ShutdownGracePeriod >= TimeSpan.Zero && ReleaseTimeout >= TimeSpan.Zero, "The shutdown periods must not be negative.");
+
+ // Longer than a timer can wait, these would throw only when the loop first waits, or at shutdown.
+ TimeSpan[] waits = [PollInterval, MaintenanceInterval, HeartbeatInterval, ShutdownGracePeriod, ReleaseTimeout];
+ Require(waits.All(wait => wait <= LongestWait), $"The intervals and shutdown periods must not be longer than {LongestWait.TotalDays} days.");
+ }
+
+ private static void Require(bool condition, string message)
+ {
+ if (!condition)
+ {
+ throw new FluentCmsConfigurationException(JobErrorCodes.InvalidRunnerOptions, message);
+ }
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/JobScheduler.cs b/src/FluentCMS.Core/Jobs/JobScheduler.cs
new file mode 100644
index 000000000..63f5b0cd5
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobScheduler.cs
@@ -0,0 +1,79 @@
+using System.Text.Json;
+using FluentCMS.Content;
+using FluentCMS.Core.Storage;
+using FluentCMS.Core.Storage.Entities;
+using FluentCMS.Jobs;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace FluentCMS.Core.Jobs;
+
+// Writes through a context of its own, so that enqueuing never saves what a caller has tracked on its own context.
+internal sealed class JobScheduler(
+ IServiceScopeFactory scopes,
+ JobRegistry registry,
+ RecurringJobs recurring,
+ IOperationContextAccessor operation,
+ JobRunnerOptions options,
+ TimeProvider time) : IJobScheduler
+{
+ public const int TypeMaxLength = 128;
+ public const int QueueMaxLength = 64;
+
+ public Task EnqueueAsync(TPayload payload, JobOptions? jobOptions = null, CancellationToken cancellationToken = default)
+ where TJob : class, IJob =>
+ AddAsync(payload, time.GetUtcNow(), jobOptions, cancellationToken);
+
+ public Task ScheduleAsync(TPayload payload, DateTimeOffset runAt, JobOptions? jobOptions = null, CancellationToken cancellationToken = default)
+ where TJob : class, IJob =>
+ AddAsync(payload, runAt, jobOptions, cancellationToken);
+
+ public Task AddRecurringAsync(string name, string cron, TPayload payload, string? queue = null, CancellationToken cancellationToken = default)
+ where TJob : class, IJob
+ {
+ ArgumentNullException.ThrowIfNull(payload);
+
+ return recurring.AddAsync(name, cron, registry.TypeOf(), Serialize(payload), CheckedQueue(queue), cancellationToken);
+ }
+
+ private async Task AddAsync(TPayload payload, DateTimeOffset runAt, JobOptions? jobOptions, CancellationToken cancellationToken)
+ where TJob : class, IJob
+ {
+ ArgumentNullException.ThrowIfNull(payload);
+
+ var maxAttempts = jobOptions?.MaxAttempts ?? options.DefaultMaxAttempts;
+ ArgumentOutOfRangeException.ThrowIfLessThan(maxAttempts, 1, nameof(jobOptions));
+
+ var current = operation.Current;
+ var job = new Job
+ {
+ Type = registry.TypeOf(),
+ Payload = Serialize(payload),
+ Queue = CheckedQueue(jobOptions?.Queue),
+ SiteId = (jobOptions?.Site ?? current?.Site)?.Value,
+ RunAt = runAt,
+ Priority = jobOptions?.Priority ?? 0,
+ MaxAttempts = maxAttempts,
+ CreatedAt = time.GetUtcNow(),
+ CreatedBy = current?.AuditPrincipal.Id is { } author && author != Guid.Empty ? author : null,
+ Origin = current?.Origin ?? Origin.Job,
+ };
+
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+ context.Add(job);
+ await context.SaveChangesAsync(cancellationToken);
+
+ return job.Id;
+ }
+
+ private static string Serialize(TPayload payload) => JsonSerializer.Serialize(payload, JobJson.Options);
+
+ private static string CheckedQueue(string? queue)
+ {
+ queue ??= JobRunnerOptions.DefaultQueue;
+ ArgumentException.ThrowIfNullOrEmpty(queue);
+ ArgumentOutOfRangeException.ThrowIfGreaterThan(queue.Length, QueueMaxLength, nameof(queue));
+
+ return queue;
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/JobStatus.cs b/src/FluentCMS.Core/Jobs/JobStatus.cs
new file mode 100644
index 000000000..b1f86a140
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobStatus.cs
@@ -0,0 +1,26 @@
+using FluentCMS.Core.Storage;
+
+namespace FluentCMS.Core.Jobs;
+
+public static class JobStatus
+{
+ public const string Pending = JobClaimStatus.Pending;
+
+ public const string Running = JobClaimStatus.Running;
+
+ public const string Succeeded = "succeeded";
+
+ public const string DeadLettered = "dead_lettered";
+}
+
+public static class JobAttemptOutcome
+{
+ public const string Succeeded = "succeeded";
+
+ public const string Failed = "failed";
+
+ public const string LeaseExpired = "lease_expired";
+
+ // A shutdown handed the job back; this does not use up an attempt.
+ public const string Released = "released";
+}
diff --git a/src/FluentCMS.Core/Jobs/JobStore.cs b/src/FluentCMS.Core/Jobs/JobStore.cs
new file mode 100644
index 000000000..26cc520cd
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobStore.cs
@@ -0,0 +1,245 @@
+using FluentCMS.Core.Storage;
+using FluentCMS.Core.Storage.Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Query;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+namespace FluentCMS.Core.Jobs;
+
+internal sealed record ClaimedJob(Guid Id, string Type, string Payload, string Queue, Guid? SiteId, int Attempt, int MaxAttempts, string Owner, DateTimeOffset LeaseUntil);
+
+// A worker can have lost its job to another by the time it writes, so every write after the claim is conditional on
+// still holding it, and matches no row otherwise. ExecuteUpdate, because SaveChanges uses UPDATE ... RETURNING on SQLite.
+internal sealed class JobStore(
+ IServiceScopeFactory scopes,
+ IJobClaimDialect dialect,
+ IJobEventSink events,
+ JobRunnerOptions options,
+ TimeProvider time,
+ ILogger logger)
+{
+ private const int SweepBatch = 100;
+ private const string LeaseExpiredError = "The worker stopped without finishing this attempt, and its lease expired.";
+ private const string OutOfAttemptsError = "The worker of the last attempt stopped without finishing it, and no attempts are left.";
+
+ public async Task ClaimAsync(string queue, string owner, CancellationToken cancellationToken = default)
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+
+ var now = time.GetUtcNow();
+
+ // The transaction takes the write lock on SQLite, which an idle poll must not do.
+ if (!await AnyDueAsync(context, queue, now, cancellationToken))
+ {
+ return null;
+ }
+
+ // One transaction, so that a failure here does not use up an attempt of a job that never ran.
+ await using var transaction = await context.Database.BeginTransactionAsync(cancellationToken);
+
+ var leaseUntil = now + options.LeaseDuration;
+ if (await dialect.ClaimAsync(context.Database, new JobClaim(queue, owner, now, leaseUntil), cancellationToken) is not { } id)
+ {
+ return null;
+ }
+
+ var job = await context.Set().AsNoTracking().SingleAsync(row => row.Id == id, cancellationToken);
+
+ // A row still open from an earlier attempt means that attempt's worker died.
+ var expired = await context.Set()
+ .Where(attempt => attempt.JobId == id && attempt.Attempt < job.Attempts && attempt.Outcome == null)
+ .ExecuteUpdateAsync(
+ attempt => attempt
+ .SetProperty(row => row.Outcome, JobAttemptOutcome.LeaseExpired)
+ .SetProperty(row => row.FinishedAt, now)
+ .SetProperty(row => row.Error, LeaseExpiredError),
+ cancellationToken);
+
+ context.Add(new JobAttempt { JobId = id, Attempt = job.Attempts, StartedAt = now });
+ await context.SaveChangesAsync(cancellationToken);
+
+ await transaction.CommitAsync(CancellationToken.None);
+
+ if (expired > 0)
+ {
+ await PublishAsync(job.Id, new JobFailedEvent(job.Id, job.Type, job.Queue, job.Attempts - 1, job.MaxAttempts, LeaseExpiredError));
+ }
+
+ return new ClaimedJob(job.Id, job.Type, job.Payload, job.Queue, job.SiteId, job.Attempts, job.MaxAttempts, owner, leaseUntil);
+ }
+
+ // The new end of the lease, or null when the worker no longer holds the job.
+ public async Task RenewLeaseAsync(ClaimedJob job)
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+
+ var until = time.GetUtcNow() + options.LeaseDuration;
+ var renewed = await Held(context, job).ExecuteUpdateAsync(row => row.SetProperty(column => column.LeaseUntil, until));
+
+ return renewed == 1 ? until : null;
+ }
+
+ public Task SucceedAsync(ClaimedJob job)
+ {
+ var now = time.GetUtcNow();
+ return SettleAsync(job, JobAttemptOutcome.Succeeded, error: null, now, row => row
+ .SetProperty(column => column.Status, JobStatus.Succeeded)
+ .SetProperty(column => column.CompletedAt, now)
+ .SetProperty(column => column.LastError, (string?)null)
+ .SetProperty(column => column.LeaseOwner, (string?)null)
+ .SetProperty(column => column.LeaseUntil, (DateTimeOffset?)null));
+ }
+
+ public async Task FailAsync(ClaimedJob job, string error)
+ {
+ var now = time.GetUtcNow();
+ var deadLetter = job.Attempt >= job.MaxAttempts;
+ var runAt = now + RetryBackoff.After(job.Attempt, options);
+
+ var held = await SettleAsync(job, JobAttemptOutcome.Failed, error, now, row =>
+ {
+ row.SetProperty(column => column.LastError, error)
+ .SetProperty(column => column.LeaseOwner, (string?)null)
+ .SetProperty(column => column.LeaseUntil, (DateTimeOffset?)null);
+
+ if (deadLetter)
+ {
+ row.SetProperty(column => column.Status, JobStatus.DeadLettered)
+ .SetProperty(column => column.CompletedAt, now);
+ }
+ else
+ {
+ row.SetProperty(column => column.Status, JobStatus.Pending)
+ .SetProperty(column => column.RunAt, runAt);
+ }
+ });
+
+ if (held)
+ {
+ await PublishAsync(job.Id, new JobFailedEvent(job.Id, job.Type, job.Queue, job.Attempt, job.MaxAttempts, error));
+ if (deadLetter)
+ {
+ JobLog.DeadLettered(logger, job.Id, job.Type, job.Attempt, error);
+ await PublishAsync(job.Id, new JobDeadLetteredEvent(job.Id, job.Type, job.Queue, job.Attempt, error));
+ }
+ }
+
+ return held;
+ }
+
+ // The attempt keeps its row, because attempts are numbered on, so max_attempts rises by one to give it back.
+ public Task ReleaseAsync(ClaimedJob job)
+ {
+ var now = time.GetUtcNow();
+ return SettleAsync(job, JobAttemptOutcome.Released, error: null, now, row => row
+ .SetProperty(column => column.Status, JobStatus.Pending)
+ .SetProperty(column => column.MaxAttempts, column => column.MaxAttempts + 1)
+ .SetProperty(column => column.LeaseOwner, (string?)null)
+ .SetProperty(column => column.LeaseUntil, (DateTimeOffset?)null));
+ }
+
+ // A claim requires attempts < max_attempts, so a job whose worker died on its last attempt never comes back and
+ // would stay running without this.
+ public async Task SweepAsync()
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+
+ var now = time.GetUtcNow();
+ var stuck = await Stuck(context, now).AsNoTracking().Take(SweepBatch).ToListAsync();
+
+ var swept = 0;
+ foreach (var job in stuck)
+ {
+ await using var transaction = await context.Database.BeginTransactionAsync();
+
+ var won = await Stuck(context, now).Where(row => row.Id == job.Id).ExecuteUpdateAsync(row => row
+ .SetProperty(column => column.Status, JobStatus.DeadLettered)
+ .SetProperty(column => column.CompletedAt, now)
+ .SetProperty(column => column.LastError, OutOfAttemptsError)
+ .SetProperty(column => column.LeaseOwner, (string?)null)
+ .SetProperty(column => column.LeaseUntil, (DateTimeOffset?)null));
+
+ if (won == 0)
+ {
+ continue;
+ }
+
+ await context.Set()
+ .Where(attempt => attempt.JobId == job.Id && attempt.Outcome == null)
+ .ExecuteUpdateAsync(attempt => attempt
+ .SetProperty(row => row.Outcome, JobAttemptOutcome.LeaseExpired)
+ .SetProperty(row => row.FinishedAt, now)
+ .SetProperty(row => row.Error, OutOfAttemptsError));
+
+ await transaction.CommitAsync();
+
+ swept++;
+ JobLog.DeadLettered(logger, job.Id, job.Type, job.Attempts, OutOfAttemptsError);
+ await PublishAsync(job.Id, new JobFailedEvent(job.Id, job.Type, job.Queue, job.Attempts, job.MaxAttempts, OutOfAttemptsError));
+ await PublishAsync(job.Id, new JobDeadLetteredEvent(job.Id, job.Type, job.Queue, job.Attempts, OutOfAttemptsError));
+ }
+
+ return swept;
+ }
+
+ private static Task AnyDueAsync(SystemDbContext context, string queue, DateTimeOffset now, CancellationToken cancellationToken) =>
+ context.Set().AnyAsync(
+ row => row.Queue == queue
+ && (row.Status == JobStatus.Pending || row.Status == JobStatus.Running)
+ && row.RunAt <= now
+ && row.Attempts < row.MaxAttempts
+ && (row.LeaseUntil == null || row.LeaseUntil < now),
+ cancellationToken);
+
+ private static IQueryable Stuck(SystemDbContext context, DateTimeOffset now) =>
+ context.Set().Where(row =>
+ (row.Status == JobStatus.Pending || row.Status == JobStatus.Running)
+ && row.Attempts >= row.MaxAttempts
+ && (row.LeaseUntil == null || row.LeaseUntil < now));
+
+ private static IQueryable Held(SystemDbContext context, ClaimedJob job) =>
+ context.Set().Where(row =>
+ row.Id == job.Id && row.Status == JobStatus.Running && row.LeaseOwner == job.Owner && row.Attempts == job.Attempt);
+
+ private async Task SettleAsync(ClaimedJob job, string outcome, string? error, DateTimeOffset now, Action> update)
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+ await using var transaction = await context.Database.BeginTransactionAsync();
+
+ if (await Held(context, job).ExecuteUpdateAsync(update) == 0)
+ {
+ return false;
+ }
+
+ await context.Set()
+ .Where(attempt => attempt.JobId == job.Id && attempt.Attempt == job.Attempt)
+ .ExecuteUpdateAsync(attempt => attempt
+ .SetProperty(row => row.Outcome, outcome)
+ .SetProperty(row => row.FinishedAt, now)
+ .SetProperty(row => row.Error, error));
+
+ await transaction.CommitAsync();
+ return true;
+ }
+
+ private Task PublishAsync(Guid jobId, JobFailedEvent failed) => GuardAsync(jobId, () => events.PublishAsync(failed));
+
+ private Task PublishAsync(Guid jobId, JobDeadLetteredEvent deadLettered) => GuardAsync(jobId, () => events.PublishAsync(deadLettered));
+
+ private async Task GuardAsync(Guid jobId, Func publish)
+ {
+ try
+ {
+ await publish();
+ }
+ catch (Exception exception)
+ {
+ JobLog.EventSinkFailed(logger, exception, jobId);
+ }
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/JobsBuilder.cs b/src/FluentCMS.Core/Jobs/JobsBuilder.cs
new file mode 100644
index 000000000..c6ce90443
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/JobsBuilder.cs
@@ -0,0 +1,28 @@
+using FluentCMS.Jobs;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace FluentCMS.Core.Jobs;
+
+public sealed class JobsBuilder
+{
+ private readonly IServiceCollection _services;
+
+ internal JobsBuilder(IServiceCollection services)
+ {
+ _services = services;
+ }
+
+ public JobRunnerOptions Options { get; } = new();
+
+ // The type is what cms_job stores, so it must outlive renames of the class.
+ public JobsBuilder AddJob(string type)
+ where TJob : class, IJob
+ {
+ ArgumentException.ThrowIfNullOrEmpty(type);
+ ArgumentOutOfRangeException.ThrowIfGreaterThan(type.Length, JobScheduler.TypeMaxLength, nameof(type));
+
+ _services.AddTransient();
+ _services.AddSingleton(JobRegistration.Create(type));
+ return this;
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/RecurringJobs.cs b/src/FluentCMS.Core/Jobs/RecurringJobs.cs
new file mode 100644
index 000000000..8e76acbcd
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/RecurringJobs.cs
@@ -0,0 +1,161 @@
+using Cronos;
+using FluentCMS.Content;
+using FluentCMS.Core.Storage;
+using FluentCMS.Core.Storage.Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+namespace FluentCMS.Core.Jobs;
+
+internal sealed class RecurringJobs(IServiceScopeFactory scopes, JobRunnerOptions options, TimeProvider time, ILogger logger)
+{
+ private const int NameMaxLength = 128;
+
+ public async Task AddAsync(string name, string cron, string type, string payload, string queue, CancellationToken cancellationToken)
+ {
+ ArgumentException.ThrowIfNullOrEmpty(name);
+ ArgumentOutOfRangeException.ThrowIfGreaterThan(name.Length, NameMaxLength, nameof(name));
+
+ var next = NextOccurrence(ParseCron(cron), time.GetUtcNow()) ?? throw new ArgumentException($"{JobErrorCodes.InvalidCron}: the cron expression '{cron}' never occurs.", nameof(cron));
+
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+
+ if (await UpdateAsync(context, name, cron, type, payload, queue, next, cancellationToken))
+ {
+ return;
+ }
+
+ context.Add(new RecurringJob { Name = name, Cron = cron, Type = type, Payload = payload, Queue = queue, NextRunAt = next, Enabled = true });
+ try
+ {
+ await context.SaveChangesAsync(cancellationToken);
+ }
+ catch (DbUpdateException)
+ {
+ // Another instance added it at the same startup, so what is left is an update.
+ context.ChangeTracker.Clear();
+ if (!await UpdateAsync(context, name, cron, type, payload, queue, next, cancellationToken))
+ {
+ throw;
+ }
+ }
+ }
+
+ public async Task FireDueAsync(CancellationToken cancellationToken = default)
+ {
+ List due;
+ var now = time.GetUtcNow();
+
+ await using (var scope = scopes.CreateAsyncScope())
+ {
+ due = await scope.ServiceProvider.GetRequiredService().Set().AsNoTracking()
+ .Where(job => job.Enabled && job.NextRunAt <= now)
+ .ToListAsync(cancellationToken);
+ }
+
+ foreach (var job in due)
+ {
+ if (Next(job, now) is not { } next)
+ {
+ continue;
+ }
+
+ try
+ {
+ await FireAsync(job, next, now, cancellationToken);
+ }
+ catch (Exception exception) when (exception is not OperationCanceledException)
+ {
+ JobLog.RecurringFireFailed(logger, exception, job.Name);
+ }
+ }
+ }
+
+ // Only the instance that moves next_run_at on enqueues the job, so an occurrence fires once however many try.
+ private async Task FireAsync(RecurringJob due, DateTimeOffset next, DateTimeOffset now, CancellationToken cancellationToken)
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var context = scope.ServiceProvider.GetRequiredService();
+ await using var transaction = await context.Database.BeginTransactionAsync(cancellationToken);
+
+ var moved = await context.Set()
+ .Where(job => job.Name == due.Name && job.Enabled && job.NextRunAt == due.NextRunAt)
+ .ExecuteUpdateAsync(job => job.SetProperty(row => row.NextRunAt, next), cancellationToken);
+
+ if (moved == 0)
+ {
+ return;
+ }
+
+ context.Add(new Job
+ {
+ Type = due.Type,
+ Payload = due.Payload,
+ Queue = due.Queue,
+ RunAt = due.NextRunAt,
+ MaxAttempts = options.DefaultMaxAttempts,
+ CreatedAt = now,
+ Origin = Origin.Job,
+ });
+ await context.SaveChangesAsync(cancellationToken);
+ await transaction.CommitAsync(cancellationToken);
+ }
+
+ // A row can hold what AddAsync would refuse, if it was edited by hand.
+ private DateTimeOffset? Next(RecurringJob job, DateTimeOffset now)
+ {
+ try
+ {
+ if (NextOccurrence(ParseCron(job.Cron), now) is { } next)
+ {
+ return next;
+ }
+
+ JobLog.RecurringCronInvalid(logger, job.Name, "it never occurs");
+ }
+ catch (ArgumentException exception)
+ {
+ JobLog.RecurringCronInvalid(logger, job.Name, exception.Message);
+ }
+
+ return null;
+ }
+
+ private static async Task UpdateAsync(SystemDbContext context, string name, string cron, string type, string payload, string queue, DateTimeOffset next, CancellationToken cancellationToken)
+ {
+ var rows = context.Set().Where(job => job.Name == name);
+
+ // Only a changed cron moves the next run, so that a restart does not skip an occurrence.
+ await rows.Where(job => job.Cron != cron)
+ .ExecuteUpdateAsync(job => job.SetProperty(row => row.NextRunAt, next), cancellationToken);
+
+ var updated = await rows.ExecuteUpdateAsync(
+ job => job
+ .SetProperty(row => row.Cron, cron)
+ .SetProperty(row => row.Type, type)
+ .SetProperty(row => row.Payload, payload)
+ .SetProperty(row => row.Queue, queue),
+ cancellationToken);
+
+ return updated == 1;
+ }
+
+ private static DateTimeOffset? NextOccurrence(CronExpression expression, DateTimeOffset after) =>
+ expression.GetNextOccurrence(after, TimeZoneInfo.Utc);
+
+ private static CronExpression ParseCron(string cron)
+ {
+ ArgumentException.ThrowIfNullOrEmpty(cron);
+
+ try
+ {
+ return CronExpression.Parse(cron);
+ }
+ catch (CronFormatException exception)
+ {
+ throw new ArgumentException($"{JobErrorCodes.InvalidCron}: the cron expression '{cron}' is not valid: {exception.Message}", nameof(cron), exception);
+ }
+ }
+}
diff --git a/src/FluentCMS.Core/Jobs/RetryBackoff.cs b/src/FluentCMS.Core/Jobs/RetryBackoff.cs
new file mode 100644
index 000000000..f47001c8d
--- /dev/null
+++ b/src/FluentCMS.Core/Jobs/RetryBackoff.cs
@@ -0,0 +1,13 @@
+namespace FluentCMS.Core.Jobs;
+
+internal static class RetryBackoff
+{
+ public static TimeSpan After(int attempt, JobRunnerOptions options)
+ {
+ var seconds = Math.Min(
+ options.RetryBaseDelay.TotalSeconds * Math.Pow(2, Math.Max(attempt - 1, 0)),
+ options.RetryMaxDelay.TotalSeconds);
+
+ return TimeSpan.FromSeconds(seconds * (1 + (Random.Shared.NextDouble() * options.RetryJitter)));
+ }
+}
diff --git a/src/FluentCMS.Core/Storage/Entities/Job.cs b/src/FluentCMS.Core/Storage/Entities/Job.cs
new file mode 100644
index 000000000..2e2daa862
--- /dev/null
+++ b/src/FluentCMS.Core/Storage/Entities/Job.cs
@@ -0,0 +1,44 @@
+using FluentCMS.Content;
+
+namespace FluentCMS.Core.Storage.Entities;
+
+// A row of cms_job. Write it through EF: on SQLite, everything after the claim binds the id as the uppercase text EF stores.
+public sealed class Job
+{
+ public const int LastErrorMaxLength = 2000;
+
+ public Guid Id { get; set; }
+
+ public string Type { get; set; } = "";
+
+ public string Payload { get; set; } = "";
+
+ public string Queue { get; set; } = "";
+
+ public Guid? SiteId { get; set; }
+
+ public DateTimeOffset RunAt { get; set; }
+
+ public int Priority { get; set; }
+
+ public string Status { get; set; } = JobClaimStatus.Pending;
+
+ public int Attempts { get; set; }
+
+ public int MaxAttempts { get; set; }
+
+ public DateTimeOffset? LeaseUntil { get; set; }
+
+ public string? LeaseOwner { get; set; }
+
+ public string? LastError { get; set; }
+
+ public DateTimeOffset CreatedAt { get; set; }
+
+ public Guid? CreatedBy { get; set; }
+
+ // Where the job was enqueued from, not how it runs: every job runs with Origin.Job.
+ public Origin Origin { get; set; }
+
+ public DateTimeOffset? CompletedAt { get; set; }
+}
diff --git a/src/FluentCMS.Core/Storage/Entities/JobAttempt.cs b/src/FluentCMS.Core/Storage/Entities/JobAttempt.cs
new file mode 100644
index 000000000..f71e51bbc
--- /dev/null
+++ b/src/FluentCMS.Core/Storage/Entities/JobAttempt.cs
@@ -0,0 +1,18 @@
+namespace FluentCMS.Core.Storage.Entities;
+
+// A row of cms_job_attempt. Outcome stays null while the attempt runs, and after a crash until the job is next
+// claimed or swept.
+public sealed class JobAttempt
+{
+ public Guid JobId { get; set; }
+
+ public int Attempt { get; set; }
+
+ public DateTimeOffset StartedAt { get; set; }
+
+ public DateTimeOffset? FinishedAt { get; set; }
+
+ public string? Outcome { get; set; }
+
+ public string? Error { get; set; }
+}
diff --git a/src/FluentCMS.Core/Storage/Entities/RecurringJob.cs b/src/FluentCMS.Core/Storage/Entities/RecurringJob.cs
new file mode 100644
index 000000000..f1d7b4689
--- /dev/null
+++ b/src/FluentCMS.Core/Storage/Entities/RecurringJob.cs
@@ -0,0 +1,19 @@
+namespace FluentCMS.Core.Storage.Entities;
+
+public sealed class RecurringJob
+{
+ public string Name { get; set; } = "";
+
+ public string Cron { get; set; } = "";
+
+ // The name the job is registered under.
+ public string Type { get; set; } = "";
+
+ public string Payload { get; set; } = "";
+
+ public string Queue { get; set; } = "";
+
+ public DateTimeOffset NextRunAt { get; set; }
+
+ public bool Enabled { get; set; }
+}
diff --git a/src/FluentCMS.Core/Storage/StorageNames.cs b/src/FluentCMS.Core/Storage/StorageNames.cs
index 88a10c79c..c773b784e 100644
--- a/src/FluentCMS.Core/Storage/StorageNames.cs
+++ b/src/FluentCMS.Core/Storage/StorageNames.cs
@@ -15,6 +15,9 @@ internal static class StorageNames
public const string Version = "cms_version";
public const string Role = "cms_role";
public const string RolePermission = "cms_role_permission";
+ public const string Job = "cms_job";
+ public const string JobAttempt = "cms_job_attempt";
+ public const string RecurringJob = "cms_recurring_job";
public const string Id = "id";
public const string DocumentId = "document_id";
diff --git a/src/FluentCMS.Core/Storage/SystemModelBuilder.cs b/src/FluentCMS.Core/Storage/SystemModelBuilder.cs
index d9afb8cb2..4a8248f7b 100644
--- a/src/FluentCMS.Core/Storage/SystemModelBuilder.cs
+++ b/src/FluentCMS.Core/Storage/SystemModelBuilder.cs
@@ -14,9 +14,17 @@ internal sealed class SystemModelBuilder(IStorageConventions conventions)
private const int PermissionLength = 128;
private const int SiteNameLength = 128;
private const int SlugLength = 64;
+ private const int JobTypeLength = 128;
+ private const int QueueLength = 64;
+ private const int LeaseOwnerLength = 128;
+ private const int RecurringNameLength = 128;
+ private const int CronLength = 128;
public void Apply(ModelBuilder modelBuilder)
{
+ modelBuilder.Entity(MapJob);
+ modelBuilder.Entity(MapJobAttempt);
+ modelBuilder.Entity(MapRecurringJob);
modelBuilder.Entity(MapSite);
modelBuilder.Entity(MapSiteLocale);
modelBuilder.Entity(MapDocument);
@@ -146,6 +154,53 @@ private void MapRoleGrant(EntityTypeBuilder entity)
.HasDatabaseName("ix_cms_role_permission_role_permission");
}
+ private void MapJob(EntityTypeBuilder entity)
+ {
+ entity.ToTable(StorageNames.Job);
+ MapColumnNames(entity);
+ entity.Property(job => job.Id).HasValueGenerator();
+ entity.Property(job => job.Type).HasMaxLength(JobTypeLength);
+ entity.Property(job => job.Queue).HasMaxLength(QueueLength);
+ entity.Property(job => job.Status).HasMaxLength(StatusLength);
+ entity.Property(job => job.LeaseOwner).HasMaxLength(LeaseOwnerLength);
+ entity.Property(job => job.LastError).HasMaxLength(Job.LastErrorMaxLength);
+ entity.Property(job => job.Origin).HasConversion().HasMaxLength(StatusLength);
+ conventions.ConfigureJsonColumn(entity, entity.Property(job => job.Payload));
+
+ // In the order of the claim's ORDER BY, so that it seeks and does not sort, and over live jobs only, so that
+ // finished ones are never walked. The claim must repeat this filter as literals for a database to use it.
+ entity.HasIndex(job => new { job.Queue, job.Priority, job.RunAt, job.Id })
+ .IsDescending(false, true, false, false)
+ .HasFilter("status IN ('pending', 'running')")
+ .HasDatabaseName("ix_cms_job_claim");
+
+ entity.HasIndex(job => new { job.Status, job.LeaseUntil })
+ .HasDatabaseName("ix_cms_job_status_lease");
+ }
+
+ private static void MapJobAttempt(EntityTypeBuilder entity)
+ {
+ entity.ToTable(StorageNames.JobAttempt);
+ MapColumnNames(entity);
+ entity.HasKey(attempt => new { attempt.JobId, attempt.Attempt });
+ entity.Property(attempt => attempt.Outcome).HasMaxLength(StatusLength);
+ entity.Property(attempt => attempt.Error).HasMaxLength(Job.LastErrorMaxLength);
+
+ entity.HasOne().WithMany().HasForeignKey(attempt => attempt.JobId).OnDelete(DeleteBehavior.Cascade);
+ }
+
+ private void MapRecurringJob(EntityTypeBuilder entity)
+ {
+ entity.ToTable(StorageNames.RecurringJob);
+ MapColumnNames(entity);
+ entity.HasKey(job => job.Name);
+ entity.Property(job => job.Name).HasMaxLength(RecurringNameLength);
+ entity.Property(job => job.Cron).HasMaxLength(CronLength);
+ entity.Property(job => job.Type).HasMaxLength(JobTypeLength);
+ entity.Property(job => job.Queue).HasMaxLength(QueueLength);
+ conventions.ConfigureJsonColumn(entity, entity.Property(job => job.Payload));
+ }
+
private static void MapColumnNames(EntityTypeBuilder entity)
where T : class
{
diff --git a/src/FluentCMS.Data.PostgreSql/Migrations/20261102202439_AddJobs.Designer.cs b/src/FluentCMS.Data.PostgreSql/Migrations/20261102202439_AddJobs.Designer.cs
new file mode 100644
index 000000000..21c467968
--- /dev/null
+++ b/src/FluentCMS.Data.PostgreSql/Migrations/20261102202439_AddJobs.Designer.cs
@@ -0,0 +1,753 @@
+//
+using System;
+using FluentCMS.Core.Storage;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Migrations;
+using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
+using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
+
+#nullable disable
+
+namespace FluentCMS.Data.PostgreSql.Migrations
+{
+ [DbContext(typeof(SystemDbContext))]
+ [Migration("20261102202439_AddJobs")]
+ partial class AddJobs
+ {
+ ///
+ protected override void BuildTargetModel(ModelBuilder modelBuilder)
+ {
+#pragma warning disable 612, 618
+ modelBuilder
+ .HasAnnotation("ProductVersion", "10.0.14")
+ .HasAnnotation("Relational:MaxIdentifierLength", 63);
+
+ NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.ContentVersion", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("AuthorId")
+ .HasColumnType("uuid")
+ .HasColumnName("author_id");
+
+ b.Property("ClientId")
+ .HasColumnType("text")
+ .HasColumnName("client_id");
+
+ b.Property("CreatedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("created_at");
+
+ b.Property("DocumentId")
+ .HasColumnType("uuid")
+ .HasColumnName("document_id");
+
+ b.Property("Locale")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("locale");
+
+ b.Property("Message")
+ .HasColumnType("text")
+ .HasColumnName("message");
+
+ b.Property("Number")
+ .HasColumnType("integer")
+ .HasColumnName("number");
+
+ b.Property("Origin")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("origin");
+
+ b.Property("Snapshot")
+ .IsRequired()
+ .HasColumnType("jsonb")
+ .HasColumnName("snapshot");
+
+ b.Property("StatusAfter")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("status_after");
+
+ b.HasKey("Id");
+
+ b.HasIndex("DocumentId", "Locale", "Number")
+ .IsUnique()
+ .HasDatabaseName("uq_cms_version_document_locale_number");
+
+ b.ToTable("cms_version", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Document", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("CreatedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("created_at");
+
+ b.Property("CreatedBy")
+ .HasColumnType("uuid")
+ .HasColumnName("created_by");
+
+ b.Property("DeletedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("deleted_at");
+
+ b.Property("DeletedBy")
+ .HasColumnType("uuid")
+ .HasColumnName("deleted_by");
+
+ b.Property("DocumentId")
+ .HasColumnType("uuid")
+ .HasColumnName("document_id");
+
+ b.Property("DraftVersion")
+ .HasColumnType("integer")
+ .HasColumnName("draft_version");
+
+ b.Property("FirstPublishedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("first_published_at");
+
+ b.Property("Locale")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("locale");
+
+ b.Property("PublishedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("published_at");
+
+ b.Property("PublishedVersion")
+ .HasColumnType("integer")
+ .HasColumnName("published_version");
+
+ b.Property("SchedulePublishAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("schedule_publish_at");
+
+ b.Property("ScheduleUnpublishAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("schedule_unpublish_at");
+
+ b.Property("SiteId")
+ .HasColumnType("uuid")
+ .HasColumnName("site_id");
+
+ b.Property("Status")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("status");
+
+ b.Property("Type")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("type");
+
+ b.Property("UpdatedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("updated_at");
+
+ b.Property("UpdatedBy")
+ .HasColumnType("uuid")
+ .HasColumnName("updated_by");
+
+ b.HasKey("Id");
+
+ b.HasIndex("DocumentId", "Locale")
+ .IsUnique()
+ .HasDatabaseName("uq_cms_document_document_locale");
+
+ b.ToTable("cms_document", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Job", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("Attempts")
+ .HasColumnType("integer")
+ .HasColumnName("attempts");
+
+ b.Property("CompletedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("completed_at");
+
+ b.Property("CreatedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("created_at");
+
+ b.Property("CreatedBy")
+ .HasColumnType("uuid")
+ .HasColumnName("created_by");
+
+ b.Property("LastError")
+ .HasMaxLength(2000)
+ .HasColumnType("character varying(2000)")
+ .HasColumnName("last_error");
+
+ b.Property("LeaseOwner")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("lease_owner");
+
+ b.Property("LeaseUntil")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("lease_until");
+
+ b.Property("MaxAttempts")
+ .HasColumnType("integer")
+ .HasColumnName("max_attempts");
+
+ b.Property("Origin")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("origin");
+
+ b.Property("Payload")
+ .IsRequired()
+ .HasColumnType("jsonb")
+ .HasColumnName("payload");
+
+ b.Property("Priority")
+ .HasColumnType("integer")
+ .HasColumnName("priority");
+
+ b.Property("Queue")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("queue");
+
+ b.Property("RunAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("run_at");
+
+ b.Property("SiteId")
+ .HasColumnType("uuid")
+ .HasColumnName("site_id");
+
+ b.Property("Status")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("status");
+
+ b.Property("Type")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("type");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Status", "LeaseUntil")
+ .HasDatabaseName("ix_cms_job_status_lease");
+
+ b.HasIndex("Queue", "Priority", "RunAt", "Id")
+ .IsDescending(false, true, false, false)
+ .HasDatabaseName("ix_cms_job_claim")
+ .HasFilter("status IN ('pending', 'running')");
+
+ b.ToTable("cms_job", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.JobAttempt", b =>
+ {
+ b.Property("JobId")
+ .HasColumnType("uuid")
+ .HasColumnName("job_id");
+
+ b.Property("Attempt")
+ .HasColumnType("integer")
+ .HasColumnName("attempt");
+
+ b.Property("Error")
+ .HasMaxLength(2000)
+ .HasColumnType("character varying(2000)")
+ .HasColumnName("error");
+
+ b.Property("FinishedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("finished_at");
+
+ b.Property("Outcome")
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("outcome");
+
+ b.Property("StartedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("started_at");
+
+ b.HasKey("JobId", "Attempt");
+
+ b.ToTable("cms_job_attempt", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.RecurringJob", b =>
+ {
+ b.Property("Name")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("name");
+
+ b.Property("Cron")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("cron");
+
+ b.Property("Enabled")
+ .HasColumnType("boolean")
+ .HasColumnName("enabled");
+
+ b.Property("NextRunAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("next_run_at");
+
+ b.Property("Payload")
+ .IsRequired()
+ .HasColumnType("jsonb")
+ .HasColumnName("payload");
+
+ b.Property("Queue")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("queue");
+
+ b.Property("Type")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("type");
+
+ b.HasKey("Name");
+
+ b.ToTable("cms_recurring_job", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Reference", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("Field")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("field");
+
+ b.Property("Locale")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("locale");
+
+ b.Property("Position")
+ .HasColumnType("integer")
+ .HasColumnName("position");
+
+ b.Property("SourceDocumentId")
+ .HasColumnType("uuid")
+ .HasColumnName("source_document_id");
+
+ b.Property("SourceType")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("source_type");
+
+ b.Property("State")
+ .HasColumnType("smallint")
+ .HasColumnName("state");
+
+ b.Property("TargetId")
+ .HasColumnType("uuid")
+ .HasColumnName("target_id");
+
+ b.Property("TargetType")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("target_type");
+
+ b.HasKey("Id");
+
+ b.HasIndex("TargetType", "TargetId")
+ .HasDatabaseName("ix_cms_reference_target");
+
+ b.HasIndex("SourceType", "SourceDocumentId", "State", "Locale", "Field", "Position")
+ .IsUnique()
+ .HasDatabaseName("uq_cms_reference_source");
+
+ b.ToTable("cms_reference", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Role", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("BuiltIn")
+ .HasColumnType("boolean")
+ .HasColumnName("built_in");
+
+ b.Property("Description")
+ .HasMaxLength(512)
+ .HasColumnType("character varying(512)")
+ .HasColumnName("description");
+
+ b.Property("Name")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("name");
+
+ b.Property("PermissionsVersion")
+ .IsConcurrencyToken()
+ .HasColumnType("bigint")
+ .HasColumnName("permissions_version");
+
+ b.Property("SiteId")
+ .HasColumnType("uuid")
+ .HasColumnName("site_id");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Name")
+ .IsUnique()
+ .HasDatabaseName("uq_cms_role_name")
+ .HasFilter("site_id IS NULL");
+
+ b.HasIndex("SiteId", "Name")
+ .IsUnique()
+ .HasDatabaseName("uq_cms_role_site_name")
+ .HasFilter("site_id IS NOT NULL");
+
+ b.ToTable("cms_role", (string)null);
+
+ b.HasData(
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8000-000000000001"),
+ BuiltIn = true,
+ Description = "Holds every permission.",
+ Name = "Super Admin",
+ PermissionsVersion = 1L
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8000-000000000002"),
+ BuiltIn = true,
+ Description = "Creates, edits, publishes and removes any content.",
+ Name = "Editor",
+ PermissionsVersion = 1L
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8000-000000000003"),
+ BuiltIn = true,
+ Description = "Creates content, and reads, edits or removes only their own entries.",
+ Name = "Author",
+ PermissionsVersion = 1L
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8000-000000000004"),
+ BuiltIn = true,
+ Description = "The anonymous visitor. Grants here open public Delivery access.",
+ Name = "Public",
+ PermissionsVersion = 1L
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8000-000000000005"),
+ BuiltIn = true,
+ Description = "The default role of a registered member.",
+ Name = "Member",
+ PermissionsVersion = 1L
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8000-000000000006"),
+ BuiltIn = true,
+ Description = "The anonymous visitor. Grants here open public Delivery access.",
+ Name = "Public",
+ PermissionsVersion = 1L,
+ SiteId = new Guid("00000000-0000-7000-8002-000000000001")
+ });
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.RoleGrant", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("Constraints")
+ .HasColumnType("jsonb")
+ .HasColumnName("constraints");
+
+ b.Property("Permission")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("permission");
+
+ b.Property("RoleId")
+ .HasColumnType("uuid")
+ .HasColumnName("role_id");
+
+ b.HasKey("Id");
+
+ b.HasIndex("RoleId", "Permission")
+ .HasDatabaseName("ix_cms_role_permission_role_permission");
+
+ b.ToTable("cms_role_permission", (string)null);
+
+ b.HasData(
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000001"),
+ Permission = "*",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000001")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000002"),
+ Permission = "content.*.create",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000003"),
+ Permission = "content.*.read",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000004"),
+ Permission = "content.*.update",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000005"),
+ Permission = "content.*.delete",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000006"),
+ Permission = "content.*.publish",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000007"),
+ Permission = "content.*.unpublish",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000008"),
+ Permission = "content.*.restore",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000009"),
+ Permission = "schema.read",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000002")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000010"),
+ Permission = "content.*.create",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000003")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000011"),
+ Constraints = "{\"rowRule\":\"own-entries\"}",
+ Permission = "content.*.read",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000003")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000012"),
+ Constraints = "{\"rowRule\":\"own-entries\"}",
+ Permission = "content.*.update",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000003")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000013"),
+ Constraints = "{\"rowRule\":\"own-entries\"}",
+ Permission = "content.*.delete",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000003")
+ },
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8001-000000000014"),
+ Permission = "schema.read",
+ RoleId = new Guid("00000000-0000-7000-8000-000000000003")
+ });
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Site", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("CreatedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("created_at");
+
+ b.Property("DefaultLocale")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("default_locale");
+
+ b.Property("IsDefault")
+ .HasColumnType("boolean")
+ .HasColumnName("is_default");
+
+ b.Property("Name")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("name");
+
+ b.Property("Slug")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("slug");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Slug")
+ .IsUnique()
+ .HasDatabaseName("uq_cms_site_slug");
+
+ b.ToTable("cms_site", (string)null);
+
+ b.HasData(
+ new
+ {
+ Id = new Guid("00000000-0000-7000-8002-000000000001"),
+ CreatedAt = new DateTimeOffset(new DateTime(2026, 11, 2, 0, 0, 0, 0, DateTimeKind.Unspecified), new TimeSpan(0, 0, 0, 0, 0)),
+ DefaultLocale = "en",
+ IsDefault = true,
+ Name = "Default",
+ Slug = "default"
+ });
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.SiteLocale", b =>
+ {
+ b.Property("SiteId")
+ .HasColumnType("uuid")
+ .HasColumnName("site_id");
+
+ b.Property("Code")
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("code");
+
+ b.Property("Enabled")
+ .HasColumnType("boolean")
+ .HasColumnName("enabled");
+
+ b.Property("Fallbacks")
+ .HasColumnType("jsonb")
+ .HasColumnName("fallbacks");
+
+ b.Property("Position")
+ .HasColumnType("integer")
+ .HasColumnName("position");
+
+ b.HasKey("SiteId", "Code");
+
+ b.ToTable("cms_site_locale", (string)null);
+
+ b.HasData(
+ new
+ {
+ SiteId = new Guid("00000000-0000-7000-8002-000000000001"),
+ Code = "en",
+ Enabled = true,
+ Position = 0
+ });
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.JobAttempt", b =>
+ {
+ b.HasOne("FluentCMS.Core.Storage.Entities.Job", null)
+ .WithMany()
+ .HasForeignKey("JobId")
+ .OnDelete(DeleteBehavior.Cascade)
+ .IsRequired();
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Role", b =>
+ {
+ b.HasOne("FluentCMS.Core.Storage.Entities.Site", null)
+ .WithMany()
+ .HasForeignKey("SiteId")
+ .OnDelete(DeleteBehavior.Cascade);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.RoleGrant", b =>
+ {
+ b.HasOne("FluentCMS.Core.Storage.Entities.Role", null)
+ .WithMany()
+ .HasForeignKey("RoleId")
+ .OnDelete(DeleteBehavior.Cascade)
+ .IsRequired();
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.SiteLocale", b =>
+ {
+ b.HasOne("FluentCMS.Core.Storage.Entities.Site", null)
+ .WithMany()
+ .HasForeignKey("SiteId")
+ .OnDelete(DeleteBehavior.Cascade)
+ .IsRequired();
+ });
+#pragma warning restore 612, 618
+ }
+ }
+}
diff --git a/src/FluentCMS.Data.PostgreSql/Migrations/20261102202439_AddJobs.cs b/src/FluentCMS.Data.PostgreSql/Migrations/20261102202439_AddJobs.cs
new file mode 100644
index 000000000..86ec8a8fb
--- /dev/null
+++ b/src/FluentCMS.Data.PostgreSql/Migrations/20261102202439_AddJobs.cs
@@ -0,0 +1,107 @@
+//
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace FluentCMS.Data.PostgreSql.Migrations
+{
+ ///
+ public partial class AddJobs : Migration
+ {
+ ///
+ protected override void Up(MigrationBuilder migrationBuilder)
+ {
+ migrationBuilder.CreateTable(
+ name: "cms_job",
+ columns: table => new
+ {
+ id = table.Column(type: "uuid", nullable: false),
+ type = table.Column(type: "character varying(128)", maxLength: 128, nullable: false),
+ payload = table.Column(type: "jsonb", nullable: false),
+ queue = table.Column(type: "character varying(64)", maxLength: 64, nullable: false),
+ site_id = table.Column(type: "uuid", nullable: true),
+ run_at = table.Column(type: "timestamp with time zone", nullable: false),
+ priority = table.Column(type: "integer", nullable: false),
+ status = table.Column(type: "character varying(16)", maxLength: 16, nullable: false),
+ attempts = table.Column(type: "integer", nullable: false),
+ max_attempts = table.Column(type: "integer", nullable: false),
+ lease_until = table.Column(type: "timestamp with time zone", nullable: true),
+ lease_owner = table.Column(type: "character varying(128)", maxLength: 128, nullable: true),
+ last_error = table.Column(type: "character varying(2000)", maxLength: 2000, nullable: true),
+ created_at = table.Column(type: "timestamp with time zone", nullable: false),
+ created_by = table.Column(type: "uuid", nullable: true),
+ origin = table.Column(type: "character varying(16)", maxLength: 16, nullable: false),
+ completed_at = table.Column(type: "timestamp with time zone", nullable: true)
+ },
+ constraints: table =>
+ {
+ table.PrimaryKey("PK_cms_job", x => x.id);
+ });
+
+ migrationBuilder.CreateTable(
+ name: "cms_recurring_job",
+ columns: table => new
+ {
+ name = table.Column(type: "character varying(128)", maxLength: 128, nullable: false),
+ cron = table.Column(type: "character varying(128)", maxLength: 128, nullable: false),
+ type = table.Column(type: "character varying(128)", maxLength: 128, nullable: false),
+ payload = table.Column(type: "jsonb", nullable: false),
+ queue = table.Column(type: "character varying(64)", maxLength: 64, nullable: false),
+ next_run_at = table.Column(type: "timestamp with time zone", nullable: false),
+ enabled = table.Column(type: "boolean", nullable: false)
+ },
+ constraints: table =>
+ {
+ table.PrimaryKey("PK_cms_recurring_job", x => x.name);
+ });
+
+ migrationBuilder.CreateTable(
+ name: "cms_job_attempt",
+ columns: table => new
+ {
+ job_id = table.Column(type: "uuid", nullable: false),
+ attempt = table.Column(type: "integer", nullable: false),
+ started_at = table.Column(type: "timestamp with time zone", nullable: false),
+ finished_at = table.Column(type: "timestamp with time zone", nullable: true),
+ outcome = table.Column(type: "character varying(16)", maxLength: 16, nullable: true),
+ error = table.Column(type: "character varying(2000)", maxLength: 2000, nullable: true)
+ },
+ constraints: table =>
+ {
+ table.PrimaryKey("PK_cms_job_attempt", x => new { x.job_id, x.attempt });
+ table.ForeignKey(
+ name: "FK_cms_job_attempt_cms_job_job_id",
+ column: x => x.job_id,
+ principalTable: "cms_job",
+ principalColumn: "id",
+ onDelete: ReferentialAction.Cascade);
+ });
+
+ migrationBuilder.CreateIndex(
+ name: "ix_cms_job_claim",
+ table: "cms_job",
+ columns: new[] { "queue", "priority", "run_at", "id" },
+ descending: new[] { false, true, false, false },
+ filter: "status IN ('pending', 'running')");
+
+ migrationBuilder.CreateIndex(
+ name: "ix_cms_job_status_lease",
+ table: "cms_job",
+ columns: new[] { "status", "lease_until" });
+ }
+
+ ///
+ protected override void Down(MigrationBuilder migrationBuilder)
+ {
+ migrationBuilder.DropTable(
+ name: "cms_job_attempt");
+
+ migrationBuilder.DropTable(
+ name: "cms_recurring_job");
+
+ migrationBuilder.DropTable(
+ name: "cms_job");
+ }
+ }
+}
diff --git a/src/FluentCMS.Data.PostgreSql/Migrations/SystemDbContextModelSnapshot.cs b/src/FluentCMS.Data.PostgreSql/Migrations/SystemDbContextModelSnapshot.cs
index 1f3e4cbaa..ea9a86439 100644
--- a/src/FluentCMS.Data.PostgreSql/Migrations/SystemDbContextModelSnapshot.cs
+++ b/src/FluentCMS.Data.PostgreSql/Migrations/SystemDbContextModelSnapshot.cs
@@ -175,6 +175,177 @@ protected override void BuildModel(ModelBuilder modelBuilder)
b.ToTable("cms_document", (string)null);
});
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Job", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("uuid")
+ .HasColumnName("id");
+
+ b.Property("Attempts")
+ .HasColumnType("integer")
+ .HasColumnName("attempts");
+
+ b.Property("CompletedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("completed_at");
+
+ b.Property("CreatedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("created_at");
+
+ b.Property("CreatedBy")
+ .HasColumnType("uuid")
+ .HasColumnName("created_by");
+
+ b.Property("LastError")
+ .HasMaxLength(2000)
+ .HasColumnType("character varying(2000)")
+ .HasColumnName("last_error");
+
+ b.Property("LeaseOwner")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("lease_owner");
+
+ b.Property("LeaseUntil")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("lease_until");
+
+ b.Property("MaxAttempts")
+ .HasColumnType("integer")
+ .HasColumnName("max_attempts");
+
+ b.Property("Origin")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("origin");
+
+ b.Property("Payload")
+ .IsRequired()
+ .HasColumnType("jsonb")
+ .HasColumnName("payload");
+
+ b.Property("Priority")
+ .HasColumnType("integer")
+ .HasColumnName("priority");
+
+ b.Property("Queue")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("queue");
+
+ b.Property("RunAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("run_at");
+
+ b.Property("SiteId")
+ .HasColumnType("uuid")
+ .HasColumnName("site_id");
+
+ b.Property("Status")
+ .IsRequired()
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("status");
+
+ b.Property("Type")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("type");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Status", "LeaseUntil")
+ .HasDatabaseName("ix_cms_job_status_lease");
+
+ b.HasIndex("Queue", "Priority", "RunAt", "Id")
+ .IsDescending(false, true, false, false)
+ .HasDatabaseName("ix_cms_job_claim")
+ .HasFilter("status IN ('pending', 'running')");
+
+ b.ToTable("cms_job", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.JobAttempt", b =>
+ {
+ b.Property("JobId")
+ .HasColumnType("uuid")
+ .HasColumnName("job_id");
+
+ b.Property("Attempt")
+ .HasColumnType("integer")
+ .HasColumnName("attempt");
+
+ b.Property("Error")
+ .HasMaxLength(2000)
+ .HasColumnType("character varying(2000)")
+ .HasColumnName("error");
+
+ b.Property("FinishedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("finished_at");
+
+ b.Property("Outcome")
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)")
+ .HasColumnName("outcome");
+
+ b.Property("StartedAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("started_at");
+
+ b.HasKey("JobId", "Attempt");
+
+ b.ToTable("cms_job_attempt", (string)null);
+ });
+
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.RecurringJob", b =>
+ {
+ b.Property("Name")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("name");
+
+ b.Property("Cron")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("cron");
+
+ b.Property("Enabled")
+ .HasColumnType("boolean")
+ .HasColumnName("enabled");
+
+ b.Property("NextRunAt")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("next_run_at");
+
+ b.Property("Payload")
+ .IsRequired()
+ .HasColumnType("jsonb")
+ .HasColumnName("payload");
+
+ b.Property("Queue")
+ .IsRequired()
+ .HasMaxLength(64)
+ .HasColumnType("character varying(64)")
+ .HasColumnName("queue");
+
+ b.Property("Type")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)")
+ .HasColumnName("type");
+
+ b.HasKey("Name");
+
+ b.ToTable("cms_recurring_job", (string)null);
+ });
+
modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Reference", b =>
{
b.Property("Id")
@@ -539,6 +710,15 @@ protected override void BuildModel(ModelBuilder modelBuilder)
});
});
+ modelBuilder.Entity("FluentCMS.Core.Storage.Entities.JobAttempt", b =>
+ {
+ b.HasOne("FluentCMS.Core.Storage.Entities.Job", null)
+ .WithMany()
+ .HasForeignKey("JobId")
+ .OnDelete(DeleteBehavior.Cascade)
+ .IsRequired();
+ });
+
modelBuilder.Entity("FluentCMS.Core.Storage.Entities.Role", b =>
{
b.HasOne("FluentCMS.Core.Storage.Entities.Site", null)
diff --git a/src/FluentCMS.Data.PostgreSql/PostgreSqlJobClaimDialect.cs b/src/FluentCMS.Data.PostgreSql/PostgreSqlJobClaimDialect.cs
index 8825b3c40..ed21f437a 100644
--- a/src/FluentCMS.Data.PostgreSql/PostgreSqlJobClaimDialect.cs
+++ b/src/FluentCMS.Data.PostgreSql/PostgreSqlJobClaimDialect.cs
@@ -20,7 +20,7 @@ UPDATE cms_job
SELECT id
FROM cms_job
WHERE queue = {claim.Queue}
- AND status IN ({JobClaimStatus.Pending}, {JobClaimStatus.Running})
+ AND status IN ('pending', 'running')
AND run_at <= {now}
AND attempts < max_attempts
AND (lease_until IS NULL OR lease_until < {now})
diff --git a/src/FluentCMS.Data.SqlServer/Migrations/20261102202445_AddJobs.Designer.cs b/src/FluentCMS.Data.SqlServer/Migrations/20261102202445_AddJobs.Designer.cs
new file mode 100644
index 000000000..8b71c541e
--- /dev/null
+++ b/src/FluentCMS.Data.SqlServer/Migrations/20261102202445_AddJobs.Designer.cs
@@ -0,0 +1,768 @@
+//