Add the in-process event bus and the outbox

src/FluentCMS.Core/Content/ContentLog.cs

@@ -6,4 +6,7 @@ internal static partial class ContentLog
 {
     [LoggerMessage(Level = LogLevel.Error, Message = "An entry change sink failed for {Kind} of {Type} {Id} in locale '{Locale}'; the write stands.")]
     public static partial void SinkFailed(ILogger logger, Exception exception, EntryChangeKind kind, string type, Guid id, string locale);
+
+    [LoggerMessage(Level = LogLevel.Error, Message = "The events raised during a write could not be dispatched; the write stands.")]
+    public static partial void EventsFailed(ILogger logger, Exception exception);
 }

src/FluentCMS.Core/Content/ContentService.cs

@@ -5,6 +5,7 @@
 using FluentCMS.Core.Content.Hooks;
 using FluentCMS.Core.Content.Query;
 using FluentCMS.Core.Content.Validation;
+using FluentCMS.Core.Events;
 using FluentCMS.Core.Localization;
 using FluentCMS.Core.Storage;
 using FluentCMS.Core.Storage.Entities;
@@ -38,6 +39,8 @@ internal sealed class ContentService(
     ContentEntryMapper mapper,
     ResponseRichText richText,
     IEnumerable<IEntryChangeSink> sinks,
+    IEnumerable<IEntryChangeStager> stagers,
+    EventBuffer events,
     ILogger<ContentService> logger) : IContentService
 {
     private readonly List<EntryChange> _pending = [];
@@ -413,20 +416,24 @@ internal async Task<T> InTransactionAsync<T>(Func<Task<T>> work, CancellationTok
         T result;
         var transaction = await context.Database.BeginTransactionAsync(cancellationToken);
         _ownsTransaction = true;
+        events.Open();
         await using (transaction)
         {
             try
             {
                 result = await work();
+                await StageAsync([.. _pending], cancellationToken);
                 await transaction.CommitAsync(CancellationToken.None);
                 committed = true;
             }
             finally
             {
                 _ownsTransaction = false;
+                events.Close();
                 if (!committed)
                 {
                     _pending.Clear();
+                    events.Clear();
                 }
             }
         }
@@ -438,6 +445,7 @@ internal async Task<T> InTransactionAsync<T>(Func<Task<T>> work, CancellationTok
     private async Task<T> InSavepointAsync<T>(IDbContextTransaction outer, Func<Task<T>> work, CancellationToken cancellationToken)
     {
         var pending = _pending.Count;
+        var buffered = events.Mark;
         var savepoint = string.Create(CultureInfo.InvariantCulture, $"cms_content_{++_savepoints}");
         if (outer.SupportsSavepoints)
         {
@@ -448,10 +456,18 @@ private async Task<T> InSavepointAsync<T>(IDbContextTransaction outer, Func<Task
         try
         {
             result = await work();
+
+            // The end of a transaction someone else began cannot be seen, so in-process handlers are never told. The rows of
+            // the outbox handlers go in with the transaction, and commit or roll back with it.
+            if (!_ownsTransaction)
+            {
+                await StageAsync([.. _pending.Skip(pending)], cancellationToken);
+            }
         }
         catch
         {
             _pending.RemoveRange(pending, _pending.Count - pending);
+            events.Truncate(buffered);
 
             // Without a savepoint the failed write's rows cannot be undone alone, and the transaction must not be
             // able to commit them, so all of it goes. SQL Server with MARS reports none: #2711.
@@ -472,20 +488,35 @@ private async Task<T> InSavepointAsync<T>(IDbContextTransaction outer, Func<Task
             await outer.ReleaseSavepointAsync(savepoint, CancellationToken.None);
         }
 
-        // The end of a transaction someone else began cannot be seen, and an entry that may yet roll back is never told.
         if (!_ownsTransaction)
         {
             _pending.RemoveRange(pending, _pending.Count - pending);
+            events.Truncate(buffered);
         }
 
         return result;
     }
 
+    // Still inside the transaction: what the outbox handlers must not lose is written with the write, and a failure here fails it.
+    private async Task StageAsync(IReadOnlyList<EntryChange> changes, CancellationToken cancellationToken)
+    {
+        foreach (var stager in stagers)
+        {
+            await stager.StageAsync(context, changes, cancellationToken);
+        }
+
+        if (_ownsTransaction)
+        {
+            await events.StageAsync(context, cancellationToken);
+        }
+    }
+
     // The write has committed: neither a failing sink nor a cancelled caller may turn it into a failure (spec 4.11).
     private async Task PublishCommittedAsync()
     {
         List<EntryChange> changes = [.. _pending];
         _pending.Clear();
+        var buffered = events.Take();
         foreach (var change in changes)
         {
             foreach (var sink in sinks)
@@ -500,6 +531,15 @@ private async Task PublishCommittedAsync()
                 }
             }
         }
+
+        try
+        {
+            await events.DispatchAsync(buffered);
+        }
+        catch (Exception exception)
+        {
+            ContentLog.EventsFailed(logger, exception);
+        }
     }
 
     private async Task<ContentDocument> SaveAsync(string typeName, Guid id, ContentDocument changes, int expectedVersion, bool isPatch, CancellationToken cancellationToken)

src/FluentCMS.Core/Content/ContentTransactionProbe.cs

@@ -0,0 +1,10 @@
+using FluentCMS.Core.Events;
+using FluentCMS.Core.Storage;
+
+namespace FluentCMS.Core.Content;
+
+// Tells the event bus whether the request is inside a transaction that nobody here controls, on either context.
+internal sealed class ContentTransactionProbe(CmsDbContext content, SystemDbContext system) : ITransactionProbe
+{
+    public bool InTransaction => content.Database.CurrentTransaction is not null || system.Database.CurrentTransaction is not null;
+}

src/FluentCMS.Core/Content/EntryChange.cs

@@ -1,4 +1,5 @@
 using FluentCMS.Content;
+using Microsoft.EntityFrameworkCore;
 
 namespace FluentCMS.Core.Content;
 
@@ -9,14 +10,26 @@ internal enum EntryChangeKind
     Restored,
 }
 
-internal sealed record EntryChange(EntryChangeKind Kind, string Type, Guid Id, string Locale, int Version, OperationContext Operation);
+internal sealed record EntryChange(EntryChangeKind Kind, string Type, Guid Id, string Locale, int Version, OperationContext Operation)
+{
+    // The id of the event the change becomes, so that the outbox row and the handlers of this process agree on it.
+    public Guid EventId { get; } = Guid.CreateVersion7();
+}
 
-// The seam for the event bus of #2697: the service tells it what changed once its transaction has committed, one change
+// The seam for the event bus: the service tells it what changed once its transaction has committed, one change
 // per locale, and tells it nothing when that rolls back. The write has happened by then, so a failure here is logged and
-// goes no further (spec 4.11). The service cannot see the end of a transaction it did not begin, so it tells nothing of
-// a write inside one: whoever needs the changes of several writes in one transaction, such as #2703, runs them in
-// ContentService.InTransactionAsync.
+// goes no further (spec 4.11). The service cannot see the end of a transaction it did not begin, so it tells this sink
+// nothing of a write inside one, and the handlers of this process hear nothing of it; whoever needs the changes of several
+// writes in one transaction, such as #2703, runs them in ContentService.InTransactionAsync. The rows for the outbox handlers
+// are written inside that transaction through IEntryChangeStager, so they commit or roll back with it and are delivered.
 internal interface IEntryChangeSink
 {
     ValueTask PublishAsync(EntryChange change, CancellationToken cancellationToken);
 }
+
+// Hears the changes of a transaction inside it, just before the commit, so that what must survive a crash can be written
+// in the same transaction. A failure here fails the write.
+internal interface IEntryChangeStager
+{
+    ValueTask StageAsync(DbContext context, IReadOnlyList<EntryChange> changes, CancellationToken cancellationToken);
+}

src/FluentCMS.Core/Content/EntryStore.cs

@@ -178,9 +178,7 @@ public Task SetTrashedAsync(Guid id, DateTimeOffset? at, Guid? by, CancellationT
     // this write's to save, or to lose if this write is rolled back, so it stays pending and only this write's rows go.
     public async Task WriteDraftAsync(SchemaType type, Document row, bool isNewDocument, ContentDocument entry, Origin origin, string? clientId, CancellationToken cancellationToken)
     {
-        context.ChangeTracker.DetectChanges();
-        var known = context.ChangeTracker.Entries().Select(item => item.Entity).ToHashSet(ReferenceEqualityComparer.Instance);
-        var held = context.ChangeTracker.Entries().Where(item => item.State != EntityState.Unchanged).Select(Hold).ToList();
+        var held = HeldChanges.Take(context);
         try
         {
             if (isNewDocument)
@@ -214,43 +212,7 @@ public async Task WriteDraftAsync(SchemaType type, Document row, bool isNewDocum
         }
         finally
         {
-            foreach (var own in context.ChangeTracker.Entries().Where(item => !known.Contains(item.Entity)).ToList())
-            {
-                own.State = EntityState.Detached;
-            }
-
-            foreach (var pending in held)
-            {
-                pending.Restore();
-            }
-        }
-    }
-
-    private static HeldEntry Hold(EntityEntry entry)
-    {
-        var original = entry.State == EntityState.Modified ? entry.OriginalValues.Clone() : null;
-        var modified = entry.Properties.Where(property => property.IsModified).Select(property => property.Metadata.Name).ToList();
-        var state = entry.State;
-        entry.State = EntityState.Detached;
-        return new HeldEntry(entry, state, original, modified);
-    }
-
-    private sealed record HeldEntry(EntityEntry Entry, EntityState State, PropertyValues? Original, List<string> Modified)
-    {
-        public void Restore()
-        {
-            if (Original is null)
-            {
-                Entry.State = State;
-                return;
-            }
-
-            Entry.State = EntityState.Unchanged;
-            Entry.OriginalValues.SetValues(Original);
-            foreach (var name in Modified)
-            {
-                Entry.Property(name).IsModified = true;
-            }
+            held.Release();
         }
     }
 

src/FluentCMS.Core/Content/HeldChanges.cs

@@ -0,0 +1,64 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.ChangeTracking;
+
+namespace FluentCMS.Core.Content;
+
+// A context may hold what its caller has not saved. That is not the save of another write to make, or to lose if that
+// write fails, so it is set aside while the write saves and put back after, and whatever the write added is detached.
+internal sealed class HeldChanges
+{
+    private readonly DbContext _context;
+    private readonly HashSet<object> _known;
+    private readonly List<HeldEntry> _held;
+
+    private HeldChanges(DbContext context)
+    {
+        _context = context;
+        context.ChangeTracker.DetectChanges();
+        _known = context.ChangeTracker.Entries().Select(item => item.Entity).ToHashSet(ReferenceEqualityComparer.Instance);
+        _held = context.ChangeTracker.Entries().Where(item => item.State != EntityState.Unchanged).Select(Hold).ToList();
+    }
+
+    public static HeldChanges Take(DbContext context) => new(context);
+
+    public void Release()
+    {
+        foreach (var own in _context.ChangeTracker.Entries().Where(item => !_known.Contains(item.Entity)).ToList())
+        {
+            own.State = EntityState.Detached;
+        }
+
+        foreach (var pending in _held)
+        {
+            pending.Restore();
+        }
+    }
+
+    private static HeldEntry Hold(EntityEntry entry)
+    {
+        var original = entry.State == EntityState.Modified ? entry.OriginalValues.Clone() : null;
+        var modified = entry.Properties.Where(property => property.IsModified).Select(property => property.Metadata.Name).ToList();
+        var state = entry.State;
+        entry.State = EntityState.Detached;
+        return new HeldEntry(entry, state, original, modified);
+    }
+
+    private sealed record HeldEntry(EntityEntry Entry, EntityState State, PropertyValues? Original, List<string> Modified)
+    {
+        public void Restore()
+        {
+            if (Original is null)
+            {
+                Entry.State = State;
+                return;
+            }
+
+            Entry.State = EntityState.Unchanged;
+            Entry.OriginalValues.SetValues(Original);
+            foreach (var name in Modified)
+            {
+                Entry.Property(name).IsModified = true;
+            }
+        }
+    }
+}

src/FluentCMS.Core/Content/OutboxStaging.cs

@@ -0,0 +1,23 @@
+using FluentCMS.Core.Storage.Entities;
+using Microsoft.EntityFrameworkCore;
+
+namespace FluentCMS.Core.Content;
+
+// Writes outbox rows in the transaction a context has open, so that they commit or roll back with it. Only these rows are
+// saved: nothing the caller tracked is, and a failed save leaves nothing behind to be saved by the next write.
+internal static class OutboxStaging
+{
+    public static async ValueTask SaveAsync(DbContext context, IReadOnlyList<OutboxMessage> rows, CancellationToken cancellationToken)
+    {
+        var held = HeldChanges.Take(context);
+        try
+        {
+            context.AddRange(rows);
+            await context.SaveChangesAsync(cancellationToken);
+        }
+        finally
+        {
+            held.Release();
+        }
+    }
+}

src/FluentCMS.Core/DependencyInjection/FluentCmsBuilder.cs

@@ -1,5 +1,6 @@
 using System.Reflection;
 using FluentCMS.Core.Authorization;
+using FluentCMS.Core.Events;
 using FluentCMS.Core.Jobs;
 using FluentCMS.Core.Storage;
 using Microsoft.Extensions.DependencyInjection;
@@ -12,6 +13,7 @@ internal FluentCmsBuilder(IServiceCollection services)
     {
         Services = services;
         Jobs = new JobsBuilder(services);
+        Events = new EventsBuilder(services);
     }
 
     public IServiceCollection Services { get; }
@@ -20,6 +22,8 @@ internal FluentCmsBuilder(IServiceCollection services)
 
     public JobsBuilder Jobs { get; }
 
+    public EventsBuilder Events { get; }
+
     internal IDataProvider? DataProvider { get; private set; }
 
     internal Assembly? ContentMigrationsAssembly { get; private set; }

src/FluentCMS.Core/DependencyInjection/FluentCmsServiceCollectionExtensions.cs

@@ -6,12 +6,14 @@
 using FluentCMS.Core.Content.Hooks;
 using FluentCMS.Core.Content.Query;
 using FluentCMS.Core.Content.Validation;
+using FluentCMS.Core.Events;
 using FluentCMS.Core.Jobs;
 using FluentCMS.Core.Localization;
 using FluentCMS.Core.ProseMirror;
 using FluentCMS.Core.Schema;
 using FluentCMS.Core.Sites;
 using FluentCMS.Core.Storage;
+using FluentCMS.Events;
 using FluentCMS.Jobs;
 using FluentCMS.Localization;
 using FluentCMS.Schema;
@@ -104,12 +106,31 @@ public static IServiceCollection AddFluentCms(this IServiceCollection services,
         services.AddSingleton<JobStore>();
         services.AddSingleton<JobExecutor>();
         services.AddSingleton<RecurringJobs>();
-        services.TryAddSingleton<IJobEventSink, NullJobEventSink>();
+        services.TryAddSingleton<IJobEventSink, BusJobEventSink>();
         services.TryAddSingleton<IJobScheduler, JobScheduler>();
         services.TryAddSingleton<IJobRunner, JobRunner>();
         services.TryAddSingleton<IJobDeadLetters, JobDeadLetters>();
         services.AddHostedService<JobRunnerHostedService>();
 
+        builder.Events.Outbox.Validate();
+        builder.Jobs.Options.Queues.TryAdd(OutboxDrainJob.Queue, 1);
+        builder.Jobs.AddJob<OutboxDrainJob, OutboxDrainPayload>(OutboxDrainJob.Type);
+        services.AddSingleton(builder.Events.Outbox);
+        services.AddSingleton<OutboxRegistry>();
+        services.AddSingleton<OutboxKick>();
+        services.AddSingleton<OutboxRows>();
+        services.TryAddSingleton<IEventDeadLetters, OutboxDeadLetters>();
+        services.AddSingleton<OutboxDrainer>();
+        services.AddHostedService<OutboxScheduleService>();
+        services.AddScoped<EventDispatcher>();
+        services.AddScoped<OutboxWriter>();
+        services.AddScoped<EventBuffer>();
+        services.AddScoped<ITransactionProbe, ContentTransactionProbe>();
+        services.TryAddScoped<IEventBus, EventBus>();
+        services.AddScoped<EntryChangeSink>();
+        services.AddScoped<IEntryChangeSink>(serviceProvider => serviceProvider.GetRequiredService<EntryChangeSink>());
+        services.AddScoped<IEntryChangeStager>(serviceProvider => serviceProvider.GetRequiredService<EntryChangeSink>());
+
         return services;
     }
 }