diff --git a/.gitignore b/.gitignore index 639c8cb..46fbec9 100644 --- a/.gitignore +++ b/.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 diff --git a/src/DigitalData.EmailProfiler.API/DigitalData.EmailProfiler.API.csproj b/src/DigitalData.EmailProfiler.API/DigitalData.EmailProfiler.API.csproj index 825ee7d..1131edd 100644 --- a/src/DigitalData.EmailProfiler.API/DigitalData.EmailProfiler.API.csproj +++ b/src/DigitalData.EmailProfiler.API/DigitalData.EmailProfiler.API.csproj @@ -9,7 +9,7 @@ - + diff --git a/src/DigitalData.EmailProfiler.API/Program.cs b/src/DigitalData.EmailProfiler.API/Program.cs index 20adb8a..40b8d56 100644 --- a/src/DigitalData.EmailProfiler.API/Program.cs +++ b/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(); builder.Services.AddControllers(); diff --git a/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs b/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs new file mode 100644 index 0000000..70fc721 --- /dev/null +++ b/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; + +/// +/// Dependency injection configuration for Infrastructure layer +/// +public static class DependencyInjection +{ + /// + /// Adds Infrastructure layer services to the DI container + /// + public static IServiceCollection AddInfrastructure( + this IServiceCollection services, + IConfiguration configuration) + { + // Register RabbitMQ configuration + services.Configure( + configuration.GetSection(RabbitMqConfiguration.SectionName)); + + // Register RabbitMQ command publisher + services.AddSingleton(); + + // Register RabbitMQ command consumer as hosted service + services.AddHostedService(); + + return services; + } +} diff --git a/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj b/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj index ef8b2fc..2d14656 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj +++ b/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj @@ -8,6 +8,13 @@ + + + + + + + diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs new file mode 100644 index 0000000..d541710 --- /dev/null +++ b/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; + +/// +/// Background service that consumes commands from RabbitMQ and executes them via MediatR +/// +public class RabbitMqCommandConsumer( + IOptions config, + IServiceProvider ServiceProvider, + ILogger? 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(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(); + + 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); + } + + /// + /// Message envelope for command deserialization + /// + 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; + } +} diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandPublisher.cs b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandPublisher.cs new file mode 100644 index 0000000..ea179d7 --- /dev/null +++ b/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; + +/// +/// RabbitMQ implementation of ICommandPublisher for asynchronous command processing +/// +public class RabbitMqCommandPublisher : ICommandPublisher, IDisposable +{ + private readonly ILogger Logger; + + private readonly RabbitMqConfiguration Config; + + private readonly IConnection Connection; + + private readonly IChannel Channel; + + private bool Disposed { get; set; } + + public RabbitMqCommandPublisher( + IOptions config, + ILogger 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; + } + } + + /// + /// Publishes a command to RabbitMQ for asynchronous processing + /// + public async Task PublishAsync(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"); + } + + /// + /// Message envelope for command serialization + /// + 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; + } +} diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs new file mode 100644 index 0000000..974a7b5 --- /dev/null +++ b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqConfiguration.cs @@ -0,0 +1,62 @@ +namespace DigitalData.EmailProfiler.Infrastructure.Messaging; + +/// +/// Configuration for RabbitMQ connection +/// +public class RabbitMqConfiguration +{ + /// + /// Configuration section name in appsettings.json + /// + public const string SectionName = "RabbitMQ"; + + /// + /// RabbitMQ server hostname + /// + public string HostName { get; set; } = "localhost"; + + /// + /// RabbitMQ AMQP port (default: 5672) + /// + public int Port { get; set; } = 5672; + + /// + /// RabbitMQ username + /// + public string UserName { get; set; } = "guest"; + + /// + /// RabbitMQ password + /// + public string Password { get; set; } = "guest"; + + /// + /// Virtual host (default: /) + /// + public string VirtualHost { get; set; } = "/"; + + /// + /// Exchange name for commands + /// + public string ExchangeName { get; set; } = "emailprofiler.commands"; + + /// + /// Queue name for commands + /// + public string QueueName { get; set; } = "emailprofiler.command.queue"; + + /// + /// Routing key for commands + /// + public string RoutingKey { get; set; } = "command"; + + /// + /// Enable automatic recovery on connection failure + /// + public bool AutomaticRecoveryEnabled { get; set; } = true; + + /// + /// Network recovery interval in seconds + /// + public int NetworkRecoveryIntervalSeconds { get; set; } = 10; +}