From 1c5bef9c2a5db66271efa5384c4042b516347893 Mon Sep 17 00:00:00 2001 From: StuartFerguson Date: Sat, 17 Oct 2026 22:14:53 +0100 Subject: [PATCH] Persist correlation IDs in event metadata --- .../AggregateRepositoryTests.cs | 42 ++++++++++++++++++ .../EventDataFactoryTests.cs | 16 +++++++ .../PersistentSubscriptionTests.cs | 43 +++++++++++++++++++ .../Aggregate/AggregateRepository.cs | 8 +++- .../Aggregate/EventDataFactory.cs | 20 ++++++++- .../Aggregate/TypeMapConvertor.cs | 8 +++- .../InMemoryPersistentSubscriptionsClient.cs | 9 +++- .../PersistentSubscription.cs | 28 ++++++++++++ 8 files changed, 167 insertions(+), 7 deletions(-) diff --git a/Shared.EventStore.Tests/AggregateRepositoryTests.cs b/Shared.EventStore.Tests/AggregateRepositoryTests.cs index cbc8a497..8c1ba25c 100644 --- 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 appendedEvents = null; + IEventStoreContextImposter context = new(); + context.InsertEvents(Arg.Any(), Arg.Any(), + Arg>.Is(events => + { + appendedEvents = events; + return true; + }), Arg.Any()) + .ReturnsAsync(Result.Success()); + + AggregateRepository 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 index 3b1c8b69..b6f04596 100644 --- 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 index fdcd1a37..8831095c 100644 --- 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.Any()).Returns(new List { 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 + { + [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())); } + + private sealed class CorrelationCapturingEventHandler : IDomainEventHandler + { + public Guid? CorrelationId { get; private set; } + + public Task 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 index d0ba59d1..628d5fe0 100644 --- 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 : StandardTraceHandler, IAggregateRepository where TAggregate : Aggregate, new() where TDomainEvent : IDomainEvent @@ -86,10 +87,13 @@ public async Task SaveChanges(TAggregate aggregate, if (!pendingEvents.Any()) return Result.Success(); + String correlationId = TenantContext.CurrentTenant?.CorrelationId.ToString(); List 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 ProcessEvents(TAggregate aggregate, } #endregion -} \ No newline at end of file +} diff --git a/Shared.EventStore/Aggregate/EventDataFactory.cs b/Shared.EventStore/Aggregate/EventDataFactory.cs index e8357f93..06a3328f 100644 --- 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? metadata = null; + if (!String.IsNullOrWhiteSpace(correlationId)) + { + Byte[] metadataBytes = JsonSerializer.SerializeToUtf8Bytes(new Dictionary + { + [TenantContext.KeyNameCorrelationId] = correlationId + }); + metadata = new ReadOnlyMemory(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) } } -} \ No newline at end of file +} diff --git a/Shared.EventStore/Aggregate/TypeMapConvertor.cs b/Shared.EventStore/Aggregate/TypeMapConvertor.cs index c2acb338..a1d551c4 100644 --- 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); + } + /// /// Convertors the specified event. /// @@ -92,4 +98,4 @@ private static IDomainEvent GetDomainEvent(Guid aggregateId, } #endregion -} \ No newline at end of file +} diff --git a/Shared.EventStore/SubscriptionWorker/InMemoryPersistentSubscriptionsClient.cs b/Shared.EventStore/SubscriptionWorker/InMemoryPersistentSubscriptionsClient.cs index be83578e..229ff404 100644 --- 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 data = new(Encoding.Default.GetBytes(@event)); IDictionary metadata = new Dictionary(); - ReadOnlyMemory custommetadata = new(Encoding.Default.GetBytes(@event)); + ReadOnlyMemory 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 -} \ No newline at end of file +} diff --git a/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs b/Shared.EventStore/SubscriptionWorker/PersistentSubscription.cs index 451114d4..83ef6a39 100644 --- 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}");