Move RavenDb id generation out of the common persistence seam

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs

@@ -196,6 +196,7 @@ public Task<QueryResult<IList<AuditCount>>> 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;

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<byte> 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)
             {

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<string, string>
+                {
+                    [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();
+        }
+    }
+}

src/ServiceControl.Audit.Persistence/ProcessedMessage.cs

@@ -19,21 +19,14 @@ public ProcessedMessage(Dictionary<string, string> headers, Dictionary<string, o
             MessageMetadata = metadata;
             Headers = headers;
 
-            var processingStartedTicks =
-                headers.TryGetValue(NServiceBus.Headers.ProcessingStarted, out var processingStartedValue)
-                    ? DateTimeOffsetHelper.ToDateTimeOffset(processingStartedValue).UtcDateTime.Ticks.ToString()
-                    : DateTime.UtcNow.Ticks.ToString();
-
-            var documentId = $"{processingStartedTicks}-{headers.ProcessingId()}";
-
-            Id = $"ProcessedMessages-{documentId}";
-
             ProcessedAt = Headers.TryGetValue(NServiceBus.Headers.ProcessingEnded, out var processedAt) ?
                 DateTimeOffsetHelper.ToDateTimeOffset(processedAt).UtcDateTime : DateTime.UtcNow; // best guess
         }
 
         public string Id { get; set; }
 
+        public string GetProcessingId() => Headers.ProcessingId();
+
         public string UniqueMessageId { get; set; }
 
         public Dictionary<string, object> MessageMetadata { get; set; }