diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs index 1ea4212771..6e4b5ed3c0 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs @@ -196,6 +196,7 @@ public Task>> QueryAuditCounts(string endpointName public Task SaveProcessedMessage(ProcessedMessage processedMessage, CancellationToken cancellationToken = default) { + processedMessage.Id ??= processedMessage.GetProcessingId(); if (processedMessages.Any(pm => pm.Id == processedMessage.Id)) { return Task.CompletedTask; diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs index 18eb78e7f7..8c21200bb9 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs @@ -11,6 +11,7 @@ using Raven.Client; using Raven.Client.Documents.BulkInsert; using Raven.Client.Json; + using ServiceControl.Audit.Persistence.Infrastructure; using ServiceControl.Infrastructure; using ServiceControl.SagaAudit; @@ -23,6 +24,11 @@ class RavenAuditIngestionUnitOfWork( { public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory body, CancellationToken cancellationToken = default) { + var processingStartedTicks = processedMessage.Headers.TryGetValue(Headers.ProcessingStarted, out var processingStartedValue) + ? DateTimeOffsetHelper.ToDateTimeOffset(processingStartedValue).UtcDateTime.Ticks + : DateTime.UtcNow.Ticks; + processedMessage.Id ??= $"ProcessedMessages-{processingStartedTicks}-{processedMessage.GetProcessingId()}"; + processedMessage.MessageMetadata["ContentLength"] = body.Length; if (!body.IsEmpty) { diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ProcessedMessageIdTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ProcessedMessageIdTests.cs new file mode 100644 index 0000000000..9d84d96eb5 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ProcessedMessageIdTests.cs @@ -0,0 +1,41 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Threading.Tasks; + using Auditing; + using NServiceBus; + using NUnit.Framework; + using ServiceControl.Audit.Infrastructure; + + [TestFixture] + class ProcessedMessageIdTests : PersistenceTestFixture + { + [Test] + public async Task Retains_existing_document_and_body_ids() + { + var started = new DateTimeOffset(2024, 1, 2, 3, 4, 5, TimeSpan.Zero); + var message = new ProcessedMessage( + new Dictionary + { + [Headers.MessageId] = "message-id", + [Headers.ProcessingEndpoint] = "endpoint", + [Headers.ProcessingStarted] = DateTimeOffsetHelper.ToWireFormattedString(started) + }, + []); + + Assert.That(message.Id, Is.Null); + + await using var unitOfWork = await StartAuditUnitOfWork(1); + await unitOfWork.RecordProcessedMessage(message, new byte[] { 42 }); + await unitOfWork.Complete(); + + var expectedId = $"ProcessedMessages-{started.UtcDateTime.Ticks}-{message.GetProcessingId()}"; + Assert.That(message.Id, Is.EqualTo(expectedId)); + Assert.That(message.MessageMetadata["BodyUrl"], Is.EqualTo($"/messages/{expectedId}/body")); + var body = await DataStore.GetMessageBody(expectedId); + Assert.That(body.HasContent, Is.True); + body.StreamContent.Dispose(); + } + } +} diff --git a/src/ServiceControl.Audit.Persistence/ProcessedMessage.cs b/src/ServiceControl.Audit.Persistence/ProcessedMessage.cs index 1082567ab9..f16bad39ed 100644 --- a/src/ServiceControl.Audit.Persistence/ProcessedMessage.cs +++ b/src/ServiceControl.Audit.Persistence/ProcessedMessage.cs @@ -19,21 +19,14 @@ public ProcessedMessage(Dictionary headers, Dictionary Headers.ProcessingId(); + public string UniqueMessageId { get; set; } public Dictionary MessageMetadata { get; set; }