StuartFerguson · Oct 18, 2026 · Oct 17, 2026
diff --git a/Shared.EventStore.Tests/AggregateRepositoryTests.cs b/Shared.EventStore.Tests/AggregateRepositoryTests.cs
@@ -14,6 +14,7 @@ namespace Shared.EventStore.Tests;
 using DomainDrivenDesign.EventSourcing;
 using EventStore;
 using global::EventStore.Client;
+using Shared.Logger.TennantContext;
 using Imposter.Abstractions;
 using Shouldly;
 using Xunit;
@@ -182,4 +183,45 @@ public async Task AggregateRepository_SaveChanges_ChangesMade_ChangesAreSaved()
         Result result = await testAggregateRepository.SaveChanges(testAggregate, CancellationToken.None);
         result.IsSuccess.ShouldBeTrue();
     }
+
+    [Fact]
+    public async Task AggregateRepository_SaveChanges_AddsSameCorrelationIdToEveryPendingEvent()
+    {
+        TenantContext previousTenantContext = TenantContext.CurrentTenant;
+        TenantContext.CurrentTenant = new TenantContext();
+        Guid correlationId = Guid.NewGuid();
+        TenantContext.CurrentTenant.SetCorrelationId(CorrelationId.From(correlationId));
+
+        try
+        {
+            List<EventData> appendedEvents = null;
+            IEventStoreContextImposter context = new();
+            context.InsertEvents(Arg<String>.Any(), Arg<long>.Any(),
+                    Arg<List<EventData>>.Is(events =>
+                    {
+                        appendedEvents = events;
+                        return true;
+                    }), Arg<CancellationToken>.Any())
+                .ReturnsAsync(Result.Success());
+
+            AggregateRepository<TestAggregate, DomainEvent> repository = new(context.Instance(), new DomainEventFactory());
+            TestAggregate aggregate = TestAggregate.Create(TestData.AggregateId);
+            aggregate.SetAggregateName("First", Guid.NewGuid());
+            aggregate.SetAggregateName("Second", Guid.NewGuid());
+
+            Result result = await repository.SaveChanges(aggregate, CancellationToken.None);
+
+            result.IsSuccess.ShouldBeTrue();
+            appendedEvents.Count.ShouldBe(2);
+            foreach (EventData eventData in appendedEvents)
+            {
+                using JsonDocument metadata = JsonDocument.Parse(eventData.Metadata);
+                metadata.RootElement.GetProperty(TenantContext.KeyNameCorrelationId).GetString().ShouldBe(correlationId.ToString("D"));
+            }
+        }
+        finally
+        {
+            TenantContext.CurrentTenant = previousTenantContext;
+        }
+    }
 }
diff --git a/Shared.EventStore.Tests/EventDataFactoryTests.cs b/Shared.EventStore.Tests/EventDataFactoryTests.cs
@@ -6,6 +6,7 @@
 using Shared.DomainDrivenDesign.EventSourcing;
 using Shared.EventStore.Aggregate;
 using Shared.EventStore.Tests.TestObjects;
+using Shared.Logger.TennantContext;
 using Shared.Serialisation;
 using Shouldly;
 using Xunit;
@@ -23,6 +24,21 @@ public void EventDataFactory_CreateEventData_EventDataCreated(){
         AggregateNameSetEvent aggregateNameSetEvent = new(TestData.AggregateId, TestData.EventId, "Test");
         EventData eventData = factory.CreateEventData(aggregateNameSetEvent);
         eventData.EventId.ToGuid().ShouldBe(aggregateNameSetEvent.EventId);
+        eventData.Metadata.IsEmpty.ShouldBeTrue();
+    }
+
+    [Fact]
+    public void EventDataFactory_CreateEventData_CorrelationIdIncludedInMetadata()
+    {
+        EventDataFactory factory = new();
+        AggregateNameSetEvent domainEvent = new(TestData.AggregateId, TestData.EventId, "Test");
+        Guid correlationId = Guid.NewGuid();
+
+        EventData eventData = factory.CreateEventData(domainEvent, correlationId.ToString("D"));
+
+        eventData.Metadata.IsEmpty.ShouldBeFalse();
+        using JsonDocument metadata = JsonDocument.Parse(eventData.Metadata);
+        metadata.RootElement.GetProperty(TenantContext.KeyNameCorrelationId).GetString().ShouldBe(correlationId.ToString("D"));
     }
 
     [Fact]

diff --git a/Shared.EventStore.Tests/PersistentSubscriptionTests.cs b/Shared.EventStore.Tests/PersistentSubscriptionTests.cs
@@ -18,6 +18,7 @@ namespace Shared.EventStore.Tests;
 using Shared.EventStore.SubscriptionWorker;
 using Shared.General;
 using Shared.Logger;
+using Shared.Logger.TennantContext;
 using Imposter.Abstractions;
 using Shouldly;
 using Xunit;
@@ -64,6 +65,37 @@ public async Task PersistentSubscription_CanBeCreatedAndReceiveEventsSingleEvent
         eventHandler.DomainEvents.Count.ShouldBe(1);
     }
 
+    [Fact]
+    public async Task PersistentSubscription_RestoresCorrelationIdFromEventMetadata()
+    {
+        TenantContext previousTenantContext = TenantContext.CurrentTenant;
+        Guid correlationId = Guid.NewGuid();
+        PersistentSubscriptionDetails details = new("$ce-test", "local-correlation");
+        CorrelationCapturingEventHandler eventHandler = new();
+        IDomainEventHandlerResolverImposter resolver = new();
+        resolver.GetDomainEventHandlers(Arg<IDomainEvent>.Any()).Returns(new List<IDomainEventHandler> { eventHandler });
+        InMemoryPersistentSubscriptionsClient client = new();
+        PersistentSubscription subscription = PersistentSubscription.Create(client, details, resolver.Instance());
+
+        try
+        {
+            await subscription.ConnectToSubscription(CancellationToken.None);
+
+            String eventJson = "{\"estateId\":\"4fc2692f-067a-443e-8006-335bf2732248\",\"estateName\":\"Demo Estate\"}";
+            String metadataJson = JsonSerializer.Serialize(new Dictionary<String, String>
+            {
+                [TenantContext.KeyNameCorrelationId] = correlationId.ToString("D")
+            });
+            client.WriteEvent(eventJson, "EstateCreatedEvent", metadataJson, CancellationToken.None);
+
+            eventHandler.CorrelationId.ShouldBe(correlationId);
+        }
+        finally
+        {
+            TenantContext.CurrentTenant = previousTenantContext;
+        }
+    }
+
     [Fact]
     public async Task PersistentSubscription_CanBeCreatedAndFilterOutSystemEvent()
     {
@@ -233,4 +265,15 @@ public void Dispose()
     {
         ConfigurationReader.Initialise(new ConfigurationRoot(new List<IConfigurationProvider>()));
     }
+
+    private sealed class CorrelationCapturingEventHandler : IDomainEventHandler
+    {
+        public Guid? CorrelationId { get; private set; }
+
+        public Task<Result> Handle(IDomainEvent domainEvent, CancellationToken cancellationToken)
+        {
+            this.CorrelationId = TenantContext.CurrentTenant?.CorrelationId.Value;
+            return Task.FromResult(Result.Success());
+        }
+    }
 }
diff --git a/Shared.EventStore/Aggregate/AggregateRepository.cs b/Shared.EventStore/Aggregate/AggregateRepository.cs
@@ -14,6 +14,7 @@ namespace Shared.EventStore.Aggregate;
 using EventStore;
 using global::EventStore.Client;
 using Shared.TraceHandler;
+using Shared.Logger.TennantContext;
 
 public sealed class AggregateRepository<TAggregate, TDomainEvent> : StandardTraceHandler, IAggregateRepository<TAggregate, TDomainEvent> where TAggregate : Aggregate, new()
     where TDomainEvent : IDomainEvent
@@ -86,10 +87,13 @@ public async Task<Result> SaveChanges(TAggregate aggregate,
         if (!pendingEvents.Any())
             return Result.Success();
 
+        String correlationId = TenantContext.CurrentTenant?.CorrelationId.ToString();
         List<EventData> events = new();
 
         foreach (IDomainEvent domainEvent in pendingEvents){
-            EventData @event = TypeMapConvertor.Convertor(domainEvent);
+            EventData @event = String.IsNullOrWhiteSpace(correlationId)
+                ? TypeMapConvertor.Convertor(domainEvent)
+                : TypeMapConvertor.Convertor(domainEvent, correlationId);
 
             events.Add(@event);
         }
@@ -142,4 +146,4 @@ private Result<TAggregate> ProcessEvents(TAggregate aggregate,
     }
 
     #endregion
-}
+}
diff --git a/Shared.EventStore/Aggregate/EventDataFactory.cs b/Shared.EventStore/Aggregate/EventDataFactory.cs
@@ -7,19 +7,35 @@ namespace Shared.EventStore.Aggregate;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
+using System.Text.Json;
 using DomainDrivenDesign.EventSourcing;
+using Shared.Logger.TennantContext;
 
 public class EventDataFactory : IEventDataFactory
 {
     private SerialiserOptions SerialiserOptions = new SerialiserOptions(SerialiserPropertyFormat.CamelCase, IgnoreNullValues: true, WriteIndented: true);
 
     public EventData CreateEventData(IDomainEvent domainEvent)
+    {
+        return this.CreateEventData(domainEvent, null);
+    }
+
+    public EventData CreateEventData(IDomainEvent domainEvent, String correlationId)
     {
         this.GuardAgainstNoDomainEvent(domainEvent);    
 
         Byte[] data = Encoding.Default.GetBytes(StringSerialiser.Serialise(domainEvent, SerialiserOptions));
+        ReadOnlyMemory<byte>? metadata = null;
+        if (!String.IsNullOrWhiteSpace(correlationId))
+        {
+            Byte[] metadataBytes = JsonSerializer.SerializeToUtf8Bytes(new Dictionary<String, String>
+            {
+                [TenantContext.KeyNameCorrelationId] = correlationId
+            });
+            metadata = new ReadOnlyMemory<Byte>(metadataBytes);
+        }
 
-        EventData eventData = new(Uuid.FromGuid(domainEvent.EventId), domainEvent.EventType, data);
+        EventData eventData = new(Uuid.FromGuid(domainEvent.EventId), domainEvent.EventType, data, metadata);
 
         return eventData;
     }
@@ -37,4 +53,4 @@ private void GuardAgainstNoDomainEvent(IDomainEvent @event)
         }
     }
 
-}
+}
diff --git a/Shared.EventStore/Aggregate/TypeMapConvertor.cs b/Shared.EventStore/Aggregate/TypeMapConvertor.cs
@@ -68,6 +68,12 @@ public static EventData Convertor(IDomainEvent @event)
         return eventDataFactory.CreateEventData(@event);
     }
 
+    public static EventData Convertor(IDomainEvent @event, String correlationId)
+    {
+        EventDataFactory eventDataFactory = new();
+        return eventDataFactory.CreateEventData(@event, correlationId);
+    }
+
     /// <summary>
     /// Convertors the specified event.
     /// </summary>
@@ -92,4 +98,4 @@ private static IDomainEvent GetDomainEvent(Guid aggregateId,
     }
 
     #endregion
-}
+}
diff --git a/Shared.EventStore/SubscriptionWorker/InMemoryPersistentSubscriptionsClient.cs b/Shared.EventStore/SubscriptionWorker/InMemoryPersistentSubscriptionsClient.cs
@@ -38,11 +38,16 @@ public class InMemoryPersistentSubscriptionsClient : IPersistentSubscriptionsCli
 
 
     public void WriteEvent(String @event, String @type, CancellationToken cancellationToken)
+    {
+        this.WriteEvent(@event, @type, @event, cancellationToken);
+    }
+
+    public void WriteEvent(String @event, String @type, String customMetadata, CancellationToken cancellationToken)
     {
 
         ReadOnlyMemory<byte> data = new(Encoding.Default.GetBytes(@event));
         IDictionary<string, string> metadata = new Dictionary<String, String>();
-        ReadOnlyMemory<byte> custommetadata = new(Encoding.Default.GetBytes(@event));
+        ReadOnlyMemory<byte> custommetadata = new(Encoding.UTF8.GetBytes(customMetadata));
 
         metadata.Add("type", @type);
         metadata.Add("created", "1000");
@@ -55,4 +60,4 @@ public void WriteEvent(String @event, String @type, CancellationToken cancellati
     }
 
     #endregion
-}
+}
diff --git a/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs b/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs
@@ -13,6 +13,7 @@ namespace Shared.EventStore.SubscriptionWorker;
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
+using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
@@ -142,6 +143,28 @@ private static TenantIdentifiers GetTenantIdentifiersFromDomainEvent(IDomainEven
         }
     }
 
+    private static CorrelationId GetCorrelationIdFromEventMetadata(ResolvedEvent resolvedEvent)
+    {
+        try
+        {
+            using JsonDocument metadata = JsonDocument.Parse(resolvedEvent.Event.Metadata);
+            if (metadata.RootElement.ValueKind == JsonValueKind.Object &&
+                metadata.RootElement.TryGetProperty(TenantContext.KeyNameCorrelationId, out JsonElement correlationIdElement) &&
+                correlationIdElement.ValueKind == JsonValueKind.String &&
+                Guid.TryParse(correlationIdElement.GetString(), out Guid correlationId) &&
+                correlationId != Guid.Empty)
+            {
+                return CorrelationId.From(correlationId);
+            }
+        }
+        catch (JsonException)
+        {
+            // Older events or events with unrelated metadata can still be processed.
+        }
+
+        return null;
+    }
+
     internal async Task EventAppeared(KurrentDB.Client.PersistentSubscription persistentSubscription,
                                              ResolvedEvent resolvedEvent,
                                              Int32? retryCount,
@@ -161,6 +184,11 @@ internal async Task EventAppeared(KurrentDB.Client.PersistentSubscription persis
 
             TenantContext tenantContext = new();
             tenantContext.Initialise(tenantIdentifiers, logPerTenantEnabled);
+            CorrelationId correlationId = PersistentSubscription.GetCorrelationIdFromEventMetadata(resolvedEvent);
+            if (correlationId != null)
+            {
+                tenantContext.SetCorrelationId(correlationId);
+            }
             TenantContext.CurrentTenant = tenantContext;
 
             Logger.Logger.LogInformation($"EventAppearedFromPersistentSubscription with Event Id {resolvedEvent.Event.EventId} event type {resolvedEvent.Event.EventType}");