feat(infrastructure): Add RabbitMQ Command Bus integration

- Add RabbitMQ.Client 7.2.1, Microsoft.Extensions.Hosting 10.0.9
- Implement ICommandPublisher interface for async command publishing
- Create RabbitMqCommandPublisher with persistent message delivery
- Create RabbitMqCommandConsumer BackgroundService for command processing
- Add RabbitMqConfiguration with appsettings.json binding
- Configure Infrastructure DI with RabbitMQ services
- Update Program.cs to support appsettings.Secrets.json
- Server: 172.24.12.56:5672, Exchange: emailprofiler.commands

.gitignore

@@ -370,3 +370,4 @@ FodyWeavers.xsd
 /EnvelopeGenerator.Server/EnvelopeGenerator.Server/publish-output
 /EnvelopeGenerator.Server/EnvelopeGenerator.Server/tekh_softHSM_test.md
 /legacy/App
+/src/DigitalData.EmailProfiler.API/appsettings.Secrets.json

src/DigitalData.EmailProfiler.API/DigitalData.EmailProfiler.API.csproj

@@ -9,7 +9,7 @@
 
   <ItemGroup>
     <PackageReference Include="Swashbuckle.AspNetCore" Version="6.6.2" />
-	<PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
+	<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.9" />
   </ItemGroup>
 
   <ItemGroup>

src/DigitalData.EmailProfiler.API/Program.cs

@@ -1,6 +1,18 @@
 using DigitalData.EmailProfiler.API;
+using DigitalData.EmailProfiler.Application;
+using DigitalData.EmailProfiler.Infrastructure;
 
 var builder = WebApplication.CreateBuilder(args);
+
+// Add appsettings.Secrets.json for sensitive configuration (not committed to git)
+builder.Configuration.AddJsonFile("appsettings.Secrets.json", optional: true, reloadOnChange: true);
+
+// Register Application layer (MediatR, AutoMapper, FluentValidation)
+builder.Services.AddApplicationServices();
+
+// Register Infrastructure layer (RabbitMQ, Repositories, etc.)
+builder.Services.AddInfrastructure(builder.Configuration);
+
 builder.Services.AddHostedService<Worker>();
 
 builder.Services.AddControllers();

src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs

@@ -0,0 +1,32 @@
+using DigitalData.EmailProfiler.Application.Common.Interfaces;
+using DigitalData.EmailProfiler.Infrastructure.Messaging;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace DigitalData.EmailProfiler.Infrastructure;
+
+/// <summary>
+/// Dependency injection configuration for Infrastructure layer
+/// </summary>
+public static class DependencyInjection
+{
+    /// <summary>
+    /// Adds Infrastructure layer services to the DI container
+    /// </summary>
+    public static IServiceCollection AddInfrastructure(
+        this IServiceCollection services,
+        IConfiguration configuration)
+    {
+        // Register RabbitMQ configuration
+        services.Configure<RabbitMqConfiguration>(
+            configuration.GetSection(RabbitMqConfiguration.SectionName));
+
+        // Register RabbitMQ command publisher
+        services.AddSingleton<ICommandPublisher, RabbitMqCommandPublisher>();
+
+        // Register RabbitMQ command consumer as hosted service
+        services.AddHostedService<RabbitMqCommandConsumer>();
+
+        return services;
+    }
+}

src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj

@@ -8,6 +8,13 @@
 
   <ItemGroup>
     <ProjectReference Include="..\DigitalData.EmailProfiler.Domain\DigitalData.EmailProfiler.Domain.csproj" />
+    <ProjectReference Include="..\DigitalData.EmailProfiler.Application\DigitalData.EmailProfiler.Application.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.9" />
+    <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.9" />
+    <PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
   </ItemGroup>
 
 </Project>

src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs

@@ -0,0 +1,188 @@
+using System.Text;
+using System.Text.Json;
+using DigitalData.EmailProfiler.Infrastructure.Messaging;
+using MediatR;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+
+namespace DigitalData.EmailProfiler.Infrastructure.Messaging;
+
+/// <summary>
+/// Background service that consumes commands from RabbitMQ and executes them via MediatR
+/// </summary>
+public class RabbitMqCommandConsumer(
+    IOptions<RabbitMqConfiguration> config,
+    IServiceProvider ServiceProvider,
+    ILogger<RabbitMqCommandConsumer>? Logger) : BackgroundService
+{
+    private readonly RabbitMqConfiguration Config = config.Value ?? throw new ArgumentNullException(nameof(config));
+    private IConnection? Connection;
+    private IChannel? _channel;
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        Logger?.LogInformation("RabbitMQ Command Consumer starting...");
+
+        try
+        {
+            // Create connection factory
+            var factory = new ConnectionFactory
+            {
+                HostName = Config.HostName,
+                Port = Config.Port,
+                UserName = Config.UserName,
+                Password = Config.Password,
+                VirtualHost = Config.VirtualHost,
+                AutomaticRecoveryEnabled = Config.AutomaticRecoveryEnabled,
+                NetworkRecoveryInterval = TimeSpan.FromSeconds(Config.NetworkRecoveryIntervalSeconds)
+            };
+
+            // Create connection and channel
+            Connection = await factory.CreateConnectionAsync(stoppingToken);
+            _channel = await Connection.CreateChannelAsync(cancellationToken: stoppingToken);
+
+            // Set prefetch count (process one message at a time)
+            await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken: stoppingToken);
+
+            // Create async consumer
+            var consumer = new AsyncEventingBasicConsumer(_channel);
+            consumer.ReceivedAsync += async (sender, ea) =>
+            {
+                try
+                {
+                    await ProcessMessageAsync(ea, stoppingToken);
+                    await _channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken: stoppingToken);
+                }
+                catch (Exception ex)
+                {
+                    Logger?.LogError(ex, "Error processing message {MessageId}", ea.BasicProperties?.MessageId);
+                    
+                    // Reject and requeue on error (be careful with infinite loops!)
+                    await _channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: true, cancellationToken: stoppingToken);
+                }
+            };
+
+            // Start consuming
+            await _channel.BasicConsumeAsync(
+                queue: Config.QueueName,
+                autoAck: false,
+                consumer: consumer,
+                cancellationToken: stoppingToken);
+
+            Logger?.LogInformation(
+                "RabbitMQ Command Consumer started. Listening on queue: {QueueName}",
+                Config.QueueName);
+
+            // Keep running until cancellation requested
+            // Consumer will process messages in the background via event handlers
+            var tcs = new TaskCompletionSource();
+            stoppingToken.Register(() => tcs.SetResult());
+            await tcs.Task;
+        }
+        catch (OperationCanceledException)
+        {
+            Logger?.LogInformation("RabbitMQ Command Consumer is stopping due to cancellation");
+        }
+        catch (Exception ex)
+        {
+            Logger?.LogError(ex, "Fatal error in RabbitMQ Command Consumer");
+            throw;
+        }
+    }
+
+    private async Task ProcessMessageAsync(BasicDeliverEventArgs ea, CancellationToken cancellationToken)
+    {
+        var messageId = ea.BasicProperties?.MessageId ?? "unknown";
+        var commandType = ea.BasicProperties?.Type ?? "unknown";
+
+        Logger?.LogInformation(
+            "Processing command {CommandType} with MessageId {MessageId}",
+            commandType, messageId);
+
+        // Deserialize message envelope
+        var json = Encoding.UTF8.GetString(ea.Body.ToArray());
+        var envelope = JsonSerializer.Deserialize<CommandEnvelope>(json);
+
+        if (envelope == null)
+        {
+            Logger?.LogWarning("Failed to deserialize command envelope for MessageId {MessageId}", messageId);
+            return;
+        }
+
+        // Get command type from assembly
+        var type = Type.GetType(envelope.CommandType);
+        if (type == null)
+        {
+            Logger?.LogWarning(
+                "Command type {CommandType} not found in assembly for MessageId {MessageId}",
+                envelope.CommandType, messageId);
+            return;
+        }
+
+        // Deserialize command payload
+        var command = JsonSerializer.Deserialize(envelope.Payload, type);
+        if (command == null)
+        {
+            Logger?.LogWarning(
+                "Failed to deserialize command payload for MessageId {MessageId}",
+                messageId);
+            return;
+        }
+
+        // Create scope and execute command via MediatR
+        using var scope = ServiceProvider.CreateScope();
+        var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
+
+        try
+        {
+            // Send command to MediatR (fire-and-forget)
+            await mediator.Send(command, cancellationToken);
+
+            Logger?.LogInformation(
+                "Successfully processed command {CommandType} with MessageId {MessageId}",
+                commandType, messageId);
+        }
+        catch (Exception ex)
+        {
+            Logger?.LogError(
+                ex,
+                "Error executing command {CommandType} with MessageId {MessageId}",
+                commandType, messageId);
+            throw;
+        }
+    }
+
+    public override async Task StopAsync(CancellationToken cancellationToken)
+    {
+        Logger?.LogInformation("RabbitMQ Command Consumer stopping...");
+
+        if (_channel != null)
+        {
+            await _channel.CloseAsync(cancellationToken);
+            _channel.Dispose();
+        }
+
+        if (Connection != null)
+        {
+            await Connection.CloseAsync(cancellationToken);
+            Connection.Dispose();
+        }
+
+        await base.StopAsync(cancellationToken);
+    }
+
+    /// <summary>
+    /// Message envelope for command deserialization
+    /// </summary>
+    private class CommandEnvelope
+    {
+        public string CommandType { get; set; } = string.Empty;
+        public string Payload { get; set; } = string.Empty;
+        public DateTime PublishedAt { get; set; }
+        public string CorrelationId { get; set; } = string.Empty;
+    }
+}

src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandPublisher.cs

@@ -0,0 +1,158 @@
+using System.Text;
+using System.Text.Json;
+using DigitalData.EmailProfiler.Application.Common.Interfaces;
+using MediatR;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using RabbitMQ.Client;
+
+namespace DigitalData.EmailProfiler.Infrastructure.Messaging;
+
+/// <summary>
+/// RabbitMQ implementation of ICommandPublisher for asynchronous command processing
+/// </summary>
+public class RabbitMqCommandPublisher : ICommandPublisher, IDisposable
+{
+    private readonly ILogger<RabbitMqCommandPublisher> Logger;
+
+    private readonly RabbitMqConfiguration Config;
+
+    private readonly IConnection Connection;
+
+    private readonly IChannel Channel;
+
+    private bool Disposed { get; set; }
+
+    public RabbitMqCommandPublisher(
+        IOptions<RabbitMqConfiguration> config,
+        ILogger<RabbitMqCommandPublisher> logger)
+    {
+        Config = config.Value ?? throw new ArgumentNullException(nameof(config));
+        Logger = logger ?? throw new ArgumentNullException(nameof(logger));
+
+        try
+        {
+            // Create connection factory
+            var factory = new ConnectionFactory
+            {
+                HostName = Config.HostName,
+                Port = Config.Port,
+                UserName = Config.UserName,
+                Password = Config.Password,
+                VirtualHost = Config.VirtualHost,
+                AutomaticRecoveryEnabled = Config.AutomaticRecoveryEnabled,
+                NetworkRecoveryInterval = TimeSpan.FromSeconds(Config.NetworkRecoveryIntervalSeconds)
+            };
+
+            // Create connection and channel
+            Connection = factory.CreateConnectionAsync().GetAwaiter().GetResult();
+            Channel = Connection.CreateChannelAsync().GetAwaiter().GetResult();
+
+            // Declare exchange (fanout for broadcasting commands)
+            Channel.ExchangeDeclareAsync(
+                exchange: Config.ExchangeName,
+                type: ExchangeType.Direct,
+                durable: true,
+                autoDelete: false).GetAwaiter().GetResult();
+
+            // Declare queue
+            Channel.QueueDeclareAsync(
+                queue: Config.QueueName,
+                durable: true,
+                exclusive: false,
+                autoDelete: false,
+                arguments: null).GetAwaiter().GetResult();
+
+            // Bind queue to exchange
+            Channel.QueueBindAsync(
+                queue: Config.QueueName,
+                exchange: Config.ExchangeName,
+                routingKey: Config.RoutingKey).GetAwaiter().GetResult();
+
+            Logger.LogInformation(
+                "RabbitMQ connection established: {HostName}:{Port}, Exchange: {Exchange}, Queue: {Queue}",
+                Config.HostName, Config.Port, Config.ExchangeName, Config.QueueName);
+        }
+        catch (Exception ex)
+        {
+            Logger.LogError(ex, "Failed to establish RabbitMQ connection");
+            throw;
+        }
+    }
+
+    /// <summary>
+    /// Publishes a command to RabbitMQ for asynchronous processing
+    /// </summary>
+    public async Task PublishAsync<TCommand>(TCommand command, CancellationToken cancellationToken = default)
+        where TCommand : IRequest
+    {
+        ObjectDisposedException.ThrowIf(Disposed, typeof(RabbitMqCommandPublisher));
+
+        try
+        {
+            // Create message envelope with metadata
+            var envelope = new CommandEnvelope
+            {
+                CommandType = typeof(TCommand).AssemblyQualifiedName!,
+                Payload = JsonSerializer.Serialize(command),
+                PublishedAt = DateTime.Now,
+                CorrelationId = Guid.NewGuid().ToString()
+            };
+
+            // Serialize to JSON
+            var json = JsonSerializer.Serialize(envelope);
+            var body = Encoding.UTF8.GetBytes(json);
+
+            // Set message properties
+            var properties = new BasicProperties
+            {
+                Persistent = true,
+                ContentType = "application/json",
+                MessageId = envelope.CorrelationId,
+                Timestamp = new AmqpTimestamp(DateTimeOffset.Now.ToUnixTimeSeconds()),
+                Type = typeof(TCommand).Name
+            };
+
+            // Publish to exchange
+            await Channel.BasicPublishAsync(
+                exchange: Config.ExchangeName,
+                routingKey: Config.RoutingKey,
+                mandatory: false,
+                basicProperties: properties,
+                body: body,
+                cancellationToken: cancellationToken);
+
+            Logger.LogInformation(
+                "Published command {CommandType} with CorrelationId {CorrelationId} to RabbitMQ",
+                typeof(TCommand).Name, envelope.CorrelationId);
+        }
+        catch (Exception ex)
+        {
+            Logger.LogError(ex, "Failed to publish command {CommandType} to RabbitMQ", typeof(TCommand).Name);
+            throw;
+        }
+    }
+
+    public void Dispose()
+    {
+        if (Disposed)
+            return;
+
+        Channel?.Dispose();
+        Connection?.Dispose();
+        Disposed = true;
+
+        Logger.LogInformation("RabbitMQ connection disposed");
+    }
+
+    /// <summary>
+    /// Message envelope for command serialization
+    /// </summary>
+    private class CommandEnvelope
+    {
+        public string CommandType { get; set; } = string.Empty;
+        public string Payload { get; set; } = string.Empty;
+        public DateTime PublishedAt { get; set; }
+        public string CorrelationId { get; set; } = string.Empty;
+    }
+}

src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs

@@ -0,0 +1,62 @@
+namespace DigitalData.EmailProfiler.Infrastructure.Messaging;
+
+/// <summary>
+/// Configuration for RabbitMQ connection
+/// </summary>
+public class RabbitMqConfiguration
+{
+    /// <summary>
+    /// Configuration section name in appsettings.json
+    /// </summary>
+    public const string SectionName = "RabbitMQ";
+
+    /// <summary>
+    /// RabbitMQ server hostname
+    /// </summary>
+    public string HostName { get; set; } = "localhost";
+
+    /// <summary>
+    /// RabbitMQ AMQP port (default: 5672)
+    /// </summary>
+    public int Port { get; set; } = 5672;
+
+    /// <summary>
+    /// RabbitMQ username
+    /// </summary>
+    public string UserName { get; set; } = "guest";
+
+    /// <summary>
+    /// RabbitMQ password
+    /// </summary>
+    public string Password { get; set; } = "guest";
+
+    /// <summary>
+    /// Virtual host (default: /)
+    /// </summary>
+    public string VirtualHost { get; set; } = "/";
+
+    /// <summary>
+    /// Exchange name for commands
+    /// </summary>
+    public string ExchangeName { get; set; } = "emailprofiler.commands";
+
+    /// <summary>
+    /// Queue name for commands
+    /// </summary>
+    public string QueueName { get; set; } = "emailprofiler.command.queue";
+
+    /// <summary>
+    /// Routing key for commands
+    /// </summary>
+    public string RoutingKey { get; set; } = "command";
+
+    /// <summary>
+    /// Enable automatic recovery on connection failure
+    /// </summary>
+    public bool AutomaticRecoveryEnabled { get; set; } = true;
+
+    /// <summary>
+    /// Network recovery interval in seconds
+    /// </summary>
+    public int NetworkRecoveryIntervalSeconds { get; set; } = 10;
+}