Adds compatibility mode for Azure Storage Queue

Introduces a compatibility mode to handle legacy messages and enables support for CorrelationId and Properties.

This change introduces a `CompatibilityMode` option to the `AzureStorageQueue` to support different message formats. The `Default` mode uses an envelope to wrap the message, preserving `CorrelationId` and `Properties`. The `Legacy` mode supports older messages serialized using the v11 SDK, ensuring backward compatibility during migration.

Author: niemyjski

src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs

@@ -31,7 +31,7 @@ public AzureStorageQueue(AzureStorageQueueOptions<T> options) : base(options)
             throw new ArgumentException("ConnectionString is required.");
 
         var clientOptions = new QueueClientOptions();
-        if (options.UseBase64Encoding)
+        if (options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Legacy)
             clientOptions.MessageEncoding = QueueMessageEncoding.Base64;
 
         options.ConfigureRetry?.Invoke(clientOptions.Retry);
@@ -76,10 +76,26 @@ protected override async Task<string> EnqueueImplAsync(T data, QueueEntryOptions
 
         Interlocked.Increment(ref _enqueuedCount);
 
-        // Note: CorrelationId and Properties from QueueEntryOptions are not persisted.
-        // Azure Storage Queue only supports a message body. Wrapping in an envelope would
-        // support these features but would break backward compatibility with existing messages.
-        var messageBody = new BinaryData(_serializer.SerializeToBytes(data));
+        byte[] messageBodyBytes;
+
+        if (_options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default)
+        {
+            // Wrap in envelope to preserve CorrelationId and Properties
+            var envelope = new QueueMessageEnvelope<T>
+            {
+                CorrelationId = options.CorrelationId,
+                Properties = options.Properties,
+                Data = data
+            };
+            messageBodyBytes = _serializer.SerializeToBytes(envelope);
+        }
+        else
+        {
+            // Legacy mode: serialize data directly for backward compatibility
+            messageBodyBytes = _serializer.SerializeToBytes(data);
+        }
+
+        var messageBody = new BinaryData(messageBodyBytes);
         var response = await _queueClient.Value.SendMessageAsync(
             messageBody,
             visibilityTimeout: options.DeliveryDelay,
@@ -155,9 +171,25 @@ protected override async Task<IQueueEntry<T>> DequeueImplAsync(CancellationToken
             message.MessageId, insertedOn, nowUtc, queueTime.TotalMilliseconds, linkedCancellationToken.IsCancellationRequested);
         Interlocked.Increment(ref _dequeuedCount);
 
-        // Deserialize the message body directly (no envelope wrapper for backward compatibility)
-        var data = _serializer.Deserialize<T>(message.Body.ToArray());
-        var entry = new AzureStorageQueueEntry<T>(message, data, this);
+        T data;
+        string correlationId = null;
+        IDictionary<string, string> properties = null;
+
+        if (_options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default)
+        {
+            // Unwrap envelope to extract metadata
+            var envelope = _serializer.Deserialize<QueueMessageEnvelope<T>>(message.Body.ToArray());
+            data = envelope.Data;
+            correlationId = envelope.CorrelationId;
+            properties = envelope.Properties;
+        }
+        else
+        {
+            // Legacy mode: deserialize data directly (no envelope)
+            data = _serializer.Deserialize<T>(message.Body.ToArray());
+        }
+
+        var entry = new AzureStorageQueueEntry<T>(message, correlationId, properties, data, this);
 
         await OnDequeuedAsync(entry).AnyContext();
         _logger.LogTrace("Dequeued message: {QueueEntryId}", message.MessageId);
@@ -402,3 +434,25 @@ private static AzureStorageQueueEntry<T> ToAzureEntryWithCheck(IQueueEntry<T> qu
         return azureQueueEntry;
     }
 }
+
+/// <summary>
+/// Envelope for wrapping queue data with metadata (correlation id, properties).
+/// Azure Storage Queue only supports a message body, so we serialize this entire envelope.
+/// </summary>
+internal record QueueMessageEnvelope<T> where T : class
+{
+    /// <summary>
+    /// Correlation ID for distributed tracing
+    /// </summary>
+    public string CorrelationId { get; init; }
+
+    /// <summary>
+    /// Custom properties/metadata
+    /// </summary>
+    public IDictionary<string, string> Properties { get; init; }
+
+    /// <summary>
+    /// The actual message payload
+    /// </summary>
+    public T Data { get; init; }
+}

src/Foundatio.AzureStorage/Queues/AzureStorageQueueEntry.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using Azure.Storage.Queues.Models;
 
 namespace Foundatio.Queues;
@@ -13,10 +14,16 @@ public class AzureStorageQueueEntry<T> : QueueEntry<T> where T : class
     /// </summary>
     public string PopReceipt { get; internal set; }
 
-    public AzureStorageQueueEntry(QueueMessage message, T data, IQueue<T> queue)
-        : base(message.MessageId, null, data, queue, message.InsertedOn?.UtcDateTime ?? DateTime.MinValue, (int)message.DequeueCount)
+    public AzureStorageQueueEntry(QueueMessage message, string correlationId, IDictionary<string, string> properties, T data, IQueue<T> queue)
+        : base(message.MessageId, correlationId, data, queue, message.InsertedOn?.UtcDateTime ?? DateTime.MinValue, (int)message.DequeueCount)
     {
         UnderlyingMessage = message;
         PopReceipt = message.PopReceipt;
+
+        if (properties != null)
+        {
+            foreach (var property in properties)
+                Properties[property.Key] = property.Value;
+        }
     }
 }

src/Foundatio.AzureStorage/Queues/AzureStorageQueueOptions.cs

@@ -3,6 +3,25 @@
 
 namespace Foundatio.Queues;
 
+/// <summary>
+/// Compatibility mode for Azure Storage Queue message format.
+/// </summary>
+public enum AzureStorageQueueCompatibilityMode
+{
+    /// <summary>
+    /// Default mode: Uses message envelope for full metadata support.
+    /// Supports CorrelationId and Properties. New installations should use this mode.
+    /// </summary>
+    Default = 0,
+
+    /// <summary>
+    /// Legacy mode: Raw message body with Base64 encoding for v11 SDK compatibility.
+    /// Use this for backward compatibility with existing queues that have messages
+    /// written with the v11 Microsoft.Azure.Storage.Queue SDK.
+    /// </summary>
+    Legacy = 1
+}
+
 public class AzureStorageQueueOptions<T> : SharedQueueOptions<T> where T : class
 {
     public string ConnectionString { get; set; }
@@ -20,28 +39,22 @@ public class AzureStorageQueueOptions<T> : SharedQueueOptions<T> where T : class
         TimeSpan.FromSeconds(Math.Pow(2, attempt)) + TimeSpan.FromMilliseconds(Random.Shared.Next(0, 100));
 
     /// <summary>
-    /// When true, messages are Base64-encoded for backward compatibility with the legacy
-    /// Microsoft.Azure.Storage.Queue SDK (v11). The v11 SDK encoded all messages as Base64
-    /// by default, while the v12 Azure.Storage.Queues SDK does not.
+    /// Controls message format compatibility.
+    /// Default mode uses an envelope wrapper that supports CorrelationId and Properties.
+    /// Legacy mode is provided for backward compatibility with existing queues that have messages
+    /// written with the v11 Microsoft.Azure.Storage.Queue SDK (uses Base64 encoding and raw payload).
     ///
-    /// <para><b>Default:</b> false (v12 behavior - no encoding)</para>
+    /// <para><b>Default:</b> AzureStorageQueueCompatibilityMode.Default (envelope with metadata)</para>
     ///
-    /// <para><b>When to enable:</b> Only enable this if you have existing messages in your
-    /// queue that were written using the v11 SDK and need to be read during migration.</para>
-    ///
-    /// <para><b>Migration path:</b></para>
+    /// <para><b>Migration from v11 SDK:</b></para>
     /// <list type="number">
-    ///   <item><description>Enable UseBase64Encoding=true to read existing v11 messages</description></item>
+    ///   <item><description>Set CompatibilityMode = Legacy to read existing v11 messages</description></item>
     ///   <item><description>Process all existing messages from the queue</description></item>
-    ///   <item><description>Once queue is empty, disable UseBase64Encoding (set to false)</description></item>
-    ///   <item><description>All new messages will use raw encoding (v12 default)</description></item>
+    ///   <item><description>Once queue is empty, switch to CompatibilityMode = Default</description></item>
+    ///   <item><description>All new messages will use envelope format with metadata support</description></item>
     /// </list>
-    ///
-    /// <para><b>Deprecation notice:</b> This option exists solely for migration purposes
-    /// and may be removed in a future major version. Plan to migrate away from Base64
-    /// encoding as soon as practical.</para>
     /// </summary>
-    public bool UseBase64Encoding { get; set; }
+    public AzureStorageQueueCompatibilityMode CompatibilityMode { get; set; } = AzureStorageQueueCompatibilityMode.Default;
 
     /// <summary>
     /// Optional action to configure Azure SDK retry options.
@@ -89,12 +102,12 @@ public AzureStorageQueueOptionsBuilder<T> RetryDelay(Func<int, TimeSpan> retryDe
     }
 
     /// <summary>
-    /// Enables Base64 message encoding for backward compatibility with the legacy v11 SDK.
-    /// See <see cref="AzureStorageQueueOptions{T}.UseBase64Encoding"/> for migration guidance.
+    /// Sets the message format compatibility mode.
+    /// See <see cref="AzureStorageQueueOptions{T}.CompatibilityMode"/> for migration guidance.
     /// </summary>
-    public AzureStorageQueueOptionsBuilder<T> UseBase64Encoding(bool useBase64Encoding = true)
+    public AzureStorageQueueOptionsBuilder<T> CompatibilityMode(AzureStorageQueueCompatibilityMode compatibilityMode)
     {
-        Target.UseBase64Encoding = useBase64Encoding;
+        Target.CompatibilityMode = compatibilityMode;
         return this;
     }
 

tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs

@@ -54,7 +54,7 @@ public override Task CanQueueAndDequeueWorkItemWithDelayAsync()
         return base.CanQueueAndDequeueWorkItemWithDelayAsync();
     }
 
-    [Fact(Skip = "Azure Storage Queue does not support CorrelationId or Properties - only message body is persisted")]
+    [Fact]
     public override Task CanUseQueueOptionsAsync()
     {
         return base.CanUseQueueOptionsAsync();