Feature: Azure Service Bus Backplane- #370

.gitignore

@@ -337,4 +337,5 @@ ASALocalRun/
 .localhistory/
 
 # BeatPulse healthcheck temp database
-healthchecksdb
+healthchecksdb
+

ZiggyCreatures.FusionCache.slnx

@@ -9,6 +9,7 @@
   </Folder>
   <Folder Name="/src/">
     <Project Path="src/ZiggyCreatures.FusionCache.AspNetCore.OutputCaching/ZiggyCreatures.FusionCache.AspNetCore.OutputCaching.csproj" />
+    <Project Path="src/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus.csproj" Id="c48413a2-1999-439c-b017-693fd79bfeec" />
     <Project Path="src/ZiggyCreatures.FusionCache.Backplane.Memory/ZiggyCreatures.FusionCache.Backplane.Memory.csproj" />
     <Project Path="src/ZiggyCreatures.FusionCache.Backplane.StackExchangeRedis/ZiggyCreatures.FusionCache.Backplane.StackExchangeRedis.csproj" />
     <Project Path="src/ZiggyCreatures.FusionCache.Chaos/ZiggyCreatures.FusionCache.Chaos.csproj" />
@@ -32,4 +33,11 @@
     <Project Path="tests/ZiggyCreatures.FusionCache.Simulator/ZiggyCreatures.FusionCache.Simulator.csproj" />
     <Project Path="tests/ZiggyCreatures.FusionCache.Tests/ZiggyCreatures.FusionCache.Tests.csproj" />
   </Folder>
+  <Folder Name="/tests/Aspire.Playground/">
+    <Project Path="tests/Aspire.Playground/Playground.AppHost/Playground.AppHost.csproj" />
+    <Project Path="tests/Aspire.Playground/Playground.ServiceDefaults/Playground.ServiceDefaults.csproj" />
+    <Project Path="tests/Aspire.Playground/Playground.Shared/Playground.Shared.csproj" />
+    <Project Path="tests/Aspire.Playground/WebApplication1/WebApplication1.csproj" />
+    <Project Path="tests/Aspire.Playground/WebApplication2/WebApplication2.csproj" />
+  </Folder>
 </Solution>

src/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus/.editorconfig

@@ -0,0 +1,6 @@
+# This .editorconfig applies only to the ZiggyCreatures.FusionCache project
+root = true
+
+[*.cs]
+# CA2007: Consider calling ConfigureAwait on the awaited task
+dotnet_diagnostic.CA2007.severity = warning

src/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus/AzureServiceBusBackplaneExtensions.cs

@@ -0,0 +1,142 @@
+using Azure.Core;
+using Azure.Messaging.ServiceBus;
+using Azure.Messaging.ServiceBus.Administration;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZiggyCreatures.Caching.Fusion;
+using ZiggyCreatures.Caching.Fusion.Backplane;
+using ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus;
+using ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus.AzureServiceBusWrapper;
+using ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus.Helpers;
+
+namespace Microsoft.Extensions.DependencyInjection;
+
+/// <summary>
+/// Extension methods for setting up FusionCache related services in an <see cref="IServiceCollection" />.
+/// </summary>
+public static class AzureServiceBusBackplaneExtensions
+{
+	private static AzureServiceBusBackplane BuildBackplane(IServiceProvider sp, AzureServiceBusBackplaneOptions options, string topicNameFallback)
+	{
+		var backplaneLogger = sp.GetService<ILogger<AzureServiceBusBackplane>>();
+
+		ValidateOptions(options);
+		var (client, adminClient) = CreateClients(options);
+		var topicName = AzureServiceBusHelpers.ResolveTopicName(options.TopicName, topicNameFallback);
+
+		string subscriptionName;
+		IAzureServiceBusAdminWrapper provisioner;
+
+		if (options.IsAdmin)
+		{
+			subscriptionName = options.SubscriptionName ?? AzureServiceBusHelpers.GenerateId();
+
+			var provisionerLogger = sp.GetService<ILogger<AzureServiceBusAdminWrapper>>() ?? NullLogger<AzureServiceBusAdminWrapper>.Instance;
+			provisioner = new AzureServiceBusAdminWrapper(adminClient, topicName, subscriptionName, provisionerLogger);
+		}
+		else
+		{
+			subscriptionName = options.SubscriptionName!;
+			provisioner = NoOpAzureServiceBusAdminWrapper.Instance;
+		}
+
+		var communicatorLogger = sp.GetService<ILogger<AzureServiceBusClientWrapper>>() ?? NullLogger<AzureServiceBusClientWrapper>.Instance;
+		var communicator = new AzureServiceBusClientWrapper(client, topicName, subscriptionName, communicatorLogger, options);
+
+		return new AzureServiceBusBackplane(communicator, provisioner, backplaneLogger, options.LockTimeout);
+	}
+
+	private static void ValidateOptions(AzureServiceBusBackplaneOptions options)
+	{
+		if (options.LockTimeout <= TimeSpan.Zero)
+			throw new InvalidOperationException($"{nameof(options.LockTimeout)} must be greater than zero.");
+
+		if (!options.IsAdmin && string.IsNullOrWhiteSpace(options.SubscriptionName))
+			throw new InvalidOperationException($"{nameof(options.SubscriptionName)} is required when {nameof(options.IsAdmin)} is false. It must identify a unique, externally provisioned subscription for this cache-process instance.");
+
+		ValidateAuthentication(options);
+	}
+
+	private static void ValidateAuthentication(AzureServiceBusBackplaneOptions options)
+	{
+		var hasConnectionString = !string.IsNullOrWhiteSpace(options.ConnectionString);
+		var hasNamespace = !string.IsNullOrWhiteSpace(options.FullyQualifiedNamespace);
+		var hasCredential = options.Credential is not null;
+
+		if (hasConnectionString && (hasNamespace || hasCredential))
+			throw new InvalidOperationException("Configure either ConnectionString or FullyQualifiedNamespace with Credential, not both.");
+
+		if (hasConnectionString)
+			return;
+
+		if (!hasNamespace || !hasCredential)
+			throw new InvalidOperationException("Configure either ConnectionString or both FullyQualifiedNamespace and Credential.");
+	}
+
+	private static (ServiceBusClient Client, ServiceBusAdministrationClient AdminClient) CreateClients(AzureServiceBusBackplaneOptions options)
+	{
+		ValidateAuthentication(options);
+
+		if (!string.IsNullOrWhiteSpace(options.ConnectionString))
+			return (new ServiceBusClient(options.ConnectionString), new ServiceBusAdministrationClient(options.ConnectionString));
+
+		return (
+			new ServiceBusClient(options.FullyQualifiedNamespace!, options.Credential!),
+			new ServiceBusAdministrationClient(options.FullyQualifiedNamespace!, options.Credential!)
+		);
+	}
+
+	/// <summary>
+	/// Adds an Azure Service Bus based implementation of a backplane to the <see cref="IServiceCollection" />.
+	/// </summary>
+	/// <param name="services">The <see cref="IServiceCollection" /> to add services to.</param>
+	/// <param name="setupOptionsAction">The <see cref="Action{AzureServiceBusBackplaneOptions}"/> to configure the provided <see cref="AzureServiceBusBackplaneOptions"/>.</param>
+	/// <returns>The <see cref="IServiceCollection"/> so that additional calls can be chained.</returns>
+	public static IServiceCollection AddFusionCacheAzureServiceBusBackplane(this IServiceCollection services, Action<AzureServiceBusBackplaneOptions>? setupOptionsAction = null)
+	{
+		if (services is null)
+			throw new ArgumentNullException(nameof(services));
+
+		services.AddOptions();
+
+		if (setupOptionsAction is not null)
+			services.Configure(setupOptionsAction);
+
+		services.TryAddTransient<IFusionCacheBackplane>(sp =>
+		{
+			var options = sp.GetRequiredService<IOptions<AzureServiceBusBackplaneOptions>>().Value;
+
+			return BuildBackplane(sp, options, FusionCacheOptions.DefaultCacheName);
+		});
+
+		return services;
+	}
+
+	/// <summary>
+	/// Adds an Azure Service Bus based implementation of a backplane to the <see cref="IFusionCacheBuilder" />.
+	/// </summary>
+	/// <param name="builder">The <see cref="IFusionCacheBuilder" /> to add the backplane to.</param>
+	/// <param name="setupOptionsAction">The <see cref="Action{AzureServiceBusBackplaneOptions}"/> to configure the provided <see cref="AzureServiceBusBackplaneOptions"/>.</param>
+	/// <returns>The <see cref="IFusionCacheBuilder"/> so that additional calls can be chained.</returns>
+	public static IFusionCacheBuilder WithAzureServiceBusBackplane(this IFusionCacheBuilder builder, Action<AzureServiceBusBackplaneOptions>? setupOptionsAction = null)
+	{
+		if (builder is null)
+			throw new ArgumentNullException(nameof(builder));
+
+		return builder
+			.WithBackplane(sp =>
+			{
+				var options = sp.GetService<IOptionsMonitor<AzureServiceBusBackplaneOptions>>()?.Get(builder.CacheName);
+
+				if (options is null)
+					throw new InvalidOperationException($"Unable to find a valid {nameof(AzureServiceBusBackplaneOptions)} instance for the current cache name '{builder.CacheName}'.");
+
+				setupOptionsAction?.Invoke(options);
+
+				return BuildBackplane(sp, options, builder.CacheName);
+			})
+		;
+	}
+}

src/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus/AzureServiceBusBackplaneOptions.cs

@@ -0,0 +1,56 @@
+using Azure.Core;
+
+namespace ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus;
+
+/// <summary>
+/// Represents the options available for the Azure Service Bus backplane.
+/// </summary>
+public class AzureServiceBusBackplaneOptions
+{
+	/// <summary>
+	/// The connection string used to connect to Azure Service Bus.
+	/// </summary>
+	public string? ConnectionString { get; set; }
+
+	/// <summary>
+	/// The fully qualified namespace (e.g. "mynamespace.servicebus.windows.net") used, together with <see cref="Credential"/>, to connect to Azure Service Bus via Azure Identity.
+	/// This is an alternative to <see cref="ConnectionString"/>.
+	/// </summary>
+	public string? FullyQualifiedNamespace { get; set; }
+
+	/// <summary>
+	/// The <see cref="TokenCredential"/> to use together with <see cref="FullyQualifiedNamespace"/> for Azure Identity based authentication.
+	/// </summary>
+	public TokenCredential? Credential { get; set; }
+
+	/// <summary>
+	/// The name of the Service Bus topic to use.
+	/// If <see langword="null"/> (the default), the cache name is used instead (sanitized into a valid Service Bus entity name).
+	/// Set this explicitly to use a specific topic, e.g. to share a single topic across multiple differently-named caches.
+	/// </summary>
+	public string? TopicName { get; set; }
+
+	/// <summary>
+	/// Whether this backplane instance is allowed to perform administrative operations against Azure Service Bus:
+	/// creating/deleting the topic, the per-instance subscription, and its self-message-filter rule. Defaults to <see langword="true"/>.
+	/// <br/>
+	/// Set to <see langword="false"/> for least-privilege deployments where the connection string/credential only has
+	/// Send/Listen claims, not Manage. In that case <see cref="SubscriptionName"/> must be set to an already-existing
+	/// subscription (provisioned out of band, e.g. via IaC), since one cannot be created on the fly, and it will never
+	/// be deleted either.
+	/// </summary>
+	public bool IsAdmin { get; set; } = true;
+
+	/// <summary>
+	/// The name of the Service Bus subscription to attach to.
+	/// Required when <see cref="IsAdmin"/> is <see langword="false"/> (the subscription must already exist).
+	/// When <see cref="IsAdmin"/> is <see langword="true"/> and this is left <see langword="null"/> (the default), a unique
+	/// subscription name is generated automatically for this instance.
+	/// </summary>
+	public string? SubscriptionName { get; set; }
+
+	/// <summary>
+	/// The max amount of time to wait to acquire the internal lock used to coordinate connection/subscription setup.
+	/// </summary>
+	public TimeSpan LockTimeout { get; set; } = TimeSpan.FromSeconds(5);
+}

src/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus/AzureServiceBusWrapper/Admin/AzureServiceBusAdminWrapper.cs

@@ -0,0 +1,66 @@
+using Azure.Messaging.ServiceBus;
+using Azure.Messaging.ServiceBus.Administration;
+using Microsoft.Extensions.Logging;
+using ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus.AzureServiceBusWrapper;
+
+namespace ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus;
+
+/// <summary>
+/// Creates and owns an instance subscription and its server-side self-message filter. Requires Manage permissions.
+/// </summary>
+/// <param name="serviceBusAdministrationClient">The administrative client used to create/delete the topic, subscription, and self-filter rule.</param>
+/// <param name="topicName">The name of the topic to create if missing.</param>
+/// <param name="subscriptionName">The name of the subscription to create if missing.</param>
+/// <param name="subscriptionAutoDeleteOnIdle">The auto-delete timeout to apply to the created subscription.</param>
+/// <param name="logger">The logger to use.</param>
+public class AzureServiceBusAdminWrapper(
+		ServiceBusAdministrationClient serviceBusAdministrationClient,
+		string topicName,
+		string subscriptionName,
+		ILogger<AzureServiceBusAdminWrapper> logger) : IAzureServiceBusAdminWrapper
+{
+	internal const string SelfMessageFilterRuleName = "FilterOutOwnMessages";
+
+	/// <inheritdoc/>
+	public async ValueTask EnsureTopicAsync()
+	{
+		if (await serviceBusAdministrationClient.TopicExistsAsync(topicName))
+			return;
+
+		await serviceBusAdministrationClient.CreateTopicAsync(topicName);
+	}
+
+	/// <inheritdoc/>
+	public async ValueTask EnsureSubscriptionAsync()
+	{
+		await EnsureTopicAsync();
+
+		if (!await serviceBusAdministrationClient.SubscriptionExistsAsync(topicName, subscriptionName))
+		{
+			logger.LogInformation("Creating a new topic subscription: {SubscriptionName}", subscriptionName);
+			await serviceBusAdministrationClient.CreateSubscriptionAsync(new CreateSubscriptionOptions(topicName, subscriptionName));
+		}
+
+		if (await serviceBusAdministrationClient.RuleExistsAsync(topicName, subscriptionName, SelfMessageFilterRuleName))
+			return;
+
+		var escapedSubscriptionName = subscriptionName.Replace("'", "''");
+		await serviceBusAdministrationClient.CreateRuleAsync(topicName, subscriptionName, new CreateRuleOptions(
+			SelfMessageFilterRuleName,
+			new SqlRuleFilter($"{AzureServiceBusClientWrapper.ConnectionIdApplicationPropertyName} <> '{escapedSubscriptionName}'")
+		));
+	}
+
+	/// <inheritdoc/>
+	public async ValueTask DisposeAsync()
+	{
+		try
+		{
+			await serviceBusAdministrationClient.DeleteSubscriptionAsync(topicName, subscriptionName);
+		}
+		catch (Exception exc)
+		{
+			logger.LogError(exc, "An error occurred while deleting subscription {SubscriptionName}", subscriptionName);
+		}
+	}
+}

src/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus/AzureServiceBusWrapper/Admin/IAzureServiceBusAdminWrapper.cs

@@ -0,0 +1,20 @@
+namespace ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus.AzureServiceBusWrapper;
+
+/// <summary>
+/// Abstracts the administrative operations (creating/deleting a topic, subscription, and self-message-filter rule) that
+/// <see cref="AzureServiceBusBackplane"/> orchestrates around an <see cref="IAzureServiceBusClientWrapper"/>: it always
+/// ensures the topic and subscription before asking the communicator to subscribe, and tears down whatever it provisioned
+/// after asking the communicator to unsubscribe.
+/// </summary>
+public interface IAzureServiceBusAdminWrapper :IAsyncDisposable
+{
+	/// <summary>
+	/// Ensures the topic exists, creating it if missing.
+	/// </summary>
+	ValueTask EnsureTopicAsync();
+
+	/// <summary>
+	/// Ensures the subscription (and its self-message-filter rule) exists, creating it if missing.
+	/// </summary>
+	ValueTask EnsureSubscriptionAsync();
+}

src/ZiggyCreatures.FusionCache.Backplane.AzureServiceBus/AzureServiceBusWrapper/Admin/NoOpAzureServiceBusAdminWrapper.cs

@@ -0,0 +1,23 @@
+namespace ZiggyCreatures.Caching.Fusion.Backplane.AzureServiceBus.AzureServiceBusWrapper;
+
+/// <summary>
+/// A no-op <see cref="IAzureServiceBusAdminWrapper"/>, used when a backplane instance has no administrative capability: it
+/// assumes the topic and subscription already exist (provisioned out of band, e.g. via IaC) and never attempts to create,
+/// delete, or otherwise administer anything. Used as a Null Object instead of a nullable/optional provisioner dependency.
+/// </summary>
+public sealed class NoOpAzureServiceBusAdminWrapper : IAzureServiceBusAdminWrapper
+{
+	/// <summary>
+	/// A shared, stateless instance.
+	/// </summary>
+	public static readonly NoOpAzureServiceBusAdminWrapper Instance = new();
+
+	/// <inheritdoc/>
+	public ValueTask EnsureTopicAsync() => default;
+
+	/// <inheritdoc/>
+	public ValueTask EnsureSubscriptionAsync() => default;
+
+	/// <inheritdoc/>
+	public ValueTask DisposeAsync() => default;
+}