move BeforeSaveChanges before ProcessEvents

src/EventSourcingTests/Aggregation/when_inline_event_changed_apply.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Marten;
+using Marten.Events.Projections;
+using Marten.Testing.Harness;
+using Shouldly;
+using Xunit;
+
+namespace EventSourcingTests.Aggregation;
+
+public class when_inline_event_enriched: OneOffConfigurationsContext
+{
+    [Fact]
+    public async Task async_apply_with_enrich_event()
+    {
+        StoreOptions(o =>
+        {
+            o.Listeners.Add(new DocumentEnricher());
+            o.Projections.Snapshot<User>(SnapshotLifecycle.Inline);
+        });
+
+        var streamId = Guid.NewGuid();
+        theSession.Events.Append(streamId, new EnrichedUser());
+        await theSession.SaveChangesAsync();
+
+        var aggregate = await theSession.Query<User>().FirstAsync();
+        aggregate.Actor.ShouldBe("some user from identity");
+    }
+
+    public class DocumentEnricher: DocumentSessionListenerBase
+    {
+        public override void BeforeProcessChanges(IDocumentSession session)
+        {
+            var streams = session.PendingChanges.Streams();
+            foreach (var stream in streams)
+            {
+                foreach (var e in stream.Events)
+                {
+                    if (e.Data is EnrichedUserEvent eueb)
+                    {
+                        eueb.Actor = "some user from identity";
+                    }
+                }
+            }
+        }
+
+        public override Task BeforeProcessChangesAsync(IDocumentSession session, CancellationToken token)
+        {
+            return Task.Factory.StartNew(() => BeforeProcessChanges(session));
+        }
+    }
+
+    public record User
+    {
+        public Guid Id { get; set; }
+        public string Actor { get; set; }
+        private void Apply(EnrichedUser e) => Actor = e.Actor;
+    }
+
+    public record EnrichedUser: EnrichedUserEvent;
+
+    public abstract record EnrichedUserEvent
+    {
+        public string Actor { get; set; }
+    }
+}

src/Marten/IDocumentSessionListener.cs

@@ -65,6 +65,22 @@ public interface IDocumentSessionListener
     /// <returns></returns>
     Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token);
 
+    /// <summary>
+    ///     Called just after IDocumentSession.SaveChanges() is called, but before
+    ///     any event apply are made
+    /// </summary>
+    /// <param name="session"></param>
+    void BeforeProcessChanges(IDocumentSession session);
+
+    /// <summary>
+    ///     Called just after IDocumentSession.SaveChanges() is called, but before
+    ///     any event apply are made
+    /// </summary>
+    /// <param name="session"></param>
+    /// <param name="token"></param>
+    /// <returns></returns>
+    Task BeforeProcessChangesAsync(IDocumentSession session, CancellationToken token);
+
     /// <summary>
     ///     Called just after IDocumentSession.SaveChanges() is called, but before
     ///     any database calls are made
@@ -107,6 +123,17 @@ public interface IDocumentSessionListener
 /// </summary>
 public abstract class DocumentSessionListenerBase: IDocumentSessionListener
 {
+    public virtual void BeforeProcessChanges(IDocumentSession session)
+    {
+        // Nothing
+    }
+
+    public virtual Task BeforeProcessChangesAsync(IDocumentSession session, CancellationToken token)
+    {
+        // Nothing
+        return Task.CompletedTask;
+    }
+
     public virtual void BeforeSaveChanges(IDocumentSession session)
     {
         // Nothing

src/Marten/Internal/Sessions/DocumentSessionBase.SaveChanges.cs

@@ -24,6 +24,8 @@ public void SaveChanges()
             return;
         }
 
+        foreach (var listener in Listeners) listener.BeforeProcessChanges(this);
+
         try
         {
             Options.EventGraph.ProcessEvents(this);
@@ -68,6 +70,11 @@ public async Task SaveChangesAsync(CancellationToken token = default)
             return;
         }
 
+        foreach (var listener in Listeners)
+        {
+            await listener.BeforeProcessChangesAsync(this, token).ConfigureAwait(false);
+        }
+
         try
         {
             await Options.EventGraph.ProcessEventsAsync(this, token).ConfigureAwait(false);
@@ -186,7 +193,6 @@ internal async Task ExecuteBatchAsync(IUpdateBatch batch, CancellationToken toke
         {
             try
             {
-
                 await executeBeforeCommitListeners(batch).ConfigureAwait(false);
 
                 await Options.ResiliencePipeline.ExecuteAsync(