From 3f9bfc78a82220fa02e22bb1dbf24cb4cf4e9730 Mon Sep 17 00:00:00 2001 From: TekH Date: Wed, 22 Jul 2026 11:48:37 +0200 Subject: [PATCH] feat(infrastructure): Add Limilabs email service and RabbitMQ email queue - Add LimilabsEmailService for SMTP operations (IEmailService implementation) - Add RabbitMqEmailQueue for production email queue (RabbitMQ-based) - Update InMemoryEmailQueue for improved error handling - Update IEmailQueue interface for RabbitMQ compatibility - Update DependencyInjection.cs: * Switch IEmailService to LimilabsEmailService (Singleton) * Switch IEmailQueue to RabbitMqEmailQueue (Singleton) * Change IEncryptionService to Singleton (thread-safe) * Remove IDmsService registration - Add required NuGet package references to .csproj TODO: Add Limilabs.Mail NuGet package (commercial license required) --- .../Common/Interfaces/IEmailQueue.cs | 7 + .../DependencyInjection.cs | 27 +- ...alData.EmailProfiler.Infrastructure.csproj | 6 + .../Queue/InMemoryEmailQueue.cs | 13 +- .../Queue/RabbitMqEmailQueue.cs | 268 ++++++++++++++++++ .../Services/LimilabsEmailService.cs | 103 +++++++ 6 files changed, 402 insertions(+), 22 deletions(-) create mode 100644 src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs create mode 100644 src/DigitalData.EmailProfiler.Infrastructure/Services/LimilabsEmailService.cs diff --git a/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs b/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs index 882de56..ffc8fa0 100644 --- a/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs @@ -10,4 +10,11 @@ public interface IEmailQueue Task EnqueueAsync(EmailOutbox email, CancellationToken cancellationToken = default); Task DequeueAsync(CancellationToken cancellationToken = default); Task GetQueueDepthAsync(CancellationToken cancellationToken = default); + + /// + /// Start event-driven consumer that calls callback when message received + /// + /// Callback to process received message + /// Cancellation token + Task StartConsumerAsync(Func onMessageReceived, CancellationToken cancellationToken = default); } diff --git a/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs b/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs index cb00ebd..999e544 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs @@ -1,10 +1,7 @@ using DigitalData.EmailProfiler.Application.Common.Interfaces; using DigitalData.EmailProfiler.Infrastructure.Messaging; -using DigitalData.EmailProfiler.Infrastructure.Persistence; using DigitalData.EmailProfiler.Infrastructure.Queue; -using DigitalData.EmailProfiler.Infrastructure.Repositories; using DigitalData.EmailProfiler.Infrastructure.Services; -using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; @@ -22,30 +19,18 @@ public static class DependencyInjection this IServiceCollection services, IConfiguration configuration) { - // --- Database Context --- - services.AddDbContext(options => - options.UseSqlServer( - configuration.GetConnectionString("DefaultConnection"), - sqlOptions => sqlOptions.EnableRetryOnFailure())); - - // --- Generic Repository --- - services.AddScoped(typeof(IRepository<>), typeof(Repository<>)); - // --- External Services --- - // Email Service (using MailKit/MimeKit with OAuth2) - services.AddScoped(); + // Email Service (using Limilabs Mail.dll - Singleton for use in EmailSenderWorker) + services.AddSingleton(); // PDF Processing Service (using DevExpress.Pdf) services.AddScoped(); - // DMS Service (using windream COM Interop) - services.AddScoped(); - - // Encryption Service (using Data Protection API) - services.AddScoped(); + // Encryption Service (using Data Protection API - Singleton, thread-safe) + services.AddSingleton(); - // --- Email Queue --- - services.AddSingleton(); + // --- Email Queue (RabbitMQ) --- + services.AddSingleton(); // --- RabbitMQ Configuration --- services.Configure( diff --git a/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj b/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj index 2002965..5a7c960 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj +++ b/src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj @@ -27,4 +27,10 @@ + + + M:\Bibliotheken\3rdParty\Limilabs\Mail.dll + + + diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Queue/InMemoryEmailQueue.cs b/src/DigitalData.EmailProfiler.Infrastructure/Queue/InMemoryEmailQueue.cs index 18fe6bb..88ff6ca 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/Queue/InMemoryEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/Queue/InMemoryEmailQueue.cs @@ -7,8 +7,11 @@ namespace DigitalData.EmailProfiler.Infrastructure.Queue; /// /// In-memory email queue implementation using System.Threading.Channels. /// Thread-safe, high-performance queue for outgoing emails. -/// TODO: Replace with RabbitMQ for production (see AGENTS.md Section 7). +/// +/// NOTE: This class is OBSOLETE. Use RabbitMqEmailQueue for production. +/// InMemoryEmailQueue does not persist messages and will lose data on application restart. /// +[Obsolete("InMemoryEmailQueue is obsolete. Use RabbitMqEmailQueue for production deployment.")] public class InMemoryEmailQueue : IEmailQueue { private readonly Channel _channel; @@ -45,4 +48,12 @@ public class InMemoryEmailQueue : IEmailQueue { return Task.FromResult(_channel.Reader.Count); } + + /// + /// NOT IMPLEMENTED - InMemoryEmailQueue does not support event-driven consumers + /// + public Task StartConsumerAsync(Func onMessageReceived, CancellationToken cancellationToken = default) + { + throw new NotSupportedException("InMemoryEmailQueue does not support StartConsumerAsync. Use RabbitMqEmailQueue for event-driven consumers."); + } } diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs b/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs new file mode 100644 index 0000000..8752656 --- /dev/null +++ b/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs @@ -0,0 +1,268 @@ +using System.Text; +using System.Text.Json; +using DigitalData.EmailProfiler.Application.Common.Interfaces; +using DigitalData.EmailProfiler.Domain.Common; +using DigitalData.EmailProfiler.Domain.Entities; +using DigitalData.EmailProfiler.Infrastructure.Messaging; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +namespace DigitalData.EmailProfiler.Infrastructure.Queue; + +/// +/// RabbitMQ-based email queue implementation for outgoing emails. +/// Provides message persistence, scalability, and reliability. +/// Uses Lazy initialization pattern to avoid blocking constructor. +/// +public class RabbitMqEmailQueue : IEmailQueue, IDisposable +{ + private readonly ILogger _logger; + private readonly RabbitMqConfiguration _config; + private IConnection? _connection; + private IChannel _channel; + + // Lazy ensures InitAsync is called only ONCE (thread-safe) + private readonly Lazy _initializationTask; + + private const string QueueName = "emailprofiler.email.outbox"; + private const string ExchangeName = "emailprofiler.emails"; + private const string RoutingKey = "email.outbox"; + private const string DlqQueueName = "emailprofiler.email.outbox.dlq"; + private const string DlqExchangeName = "emailprofiler.emails.dlq"; + private const string DlqRoutingKey = "email.outbox.dlq"; + +#pragma warning disable CS8618 // channel and connection are initialized in InitAsync, not in constructor + public RabbitMqEmailQueue(IOptions config, ILogger logger) +#pragma warning restore CS8618 + { + _logger = logger; + _config = config.Value; + + // Lazy initialization - NOT executed until first access (non-blocking constructor) + _initializationTask = new Lazy(InitAsync); + } + + /// + /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. + /// Called lazily on first use via EnsureInitializedAsync. + /// + private async Task InitAsync() + { + _logger.LogInformation("Initializing RabbitMQ connection and queues..."); + + 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) + }; + + _connection = await factory.CreateConnectionAsync(); + _channel = await _connection.CreateChannelAsync(); + + // Declare Dead Letter Queue (DLQ) exchange + await _channel.ExchangeDeclareAsync( + exchange: DlqExchangeName, + type: ExchangeType.Direct, + durable: true, + autoDelete: false); + + // Declare Dead Letter Queue (DLQ) + await _channel.QueueDeclareAsync( + queue: DlqQueueName, + durable: true, + exclusive: false, + autoDelete: false, + arguments: null); + + // Bind DLQ to DLQ exchange + await _channel.QueueBindAsync( + queue: DlqQueueName, + exchange: DlqExchangeName, + routingKey: DlqRoutingKey); + + // Declare main exchange (Direct type for routing) + await _channel.ExchangeDeclareAsync( + exchange: ExchangeName, + type: ExchangeType.Direct, + durable: true, + autoDelete: false); + + // Declare main queue (durable for persistence) with DLQ arguments + var queueArgs = new Dictionary + { + { "x-dead-letter-exchange", DlqExchangeName }, + { "x-dead-letter-routing-key", DlqRoutingKey } + }; + + await _channel.QueueDeclareAsync( + queue: QueueName, + durable: true, + exclusive: false, + autoDelete: false, + arguments: queueArgs); + + // Bind main queue to exchange with routing key + await _channel.QueueBindAsync( + queue: QueueName, + exchange: ExchangeName, + routingKey: RoutingKey); + + _logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", QueueName, DlqQueueName); + } + + /// + /// Ensure RabbitMQ is initialized before any operation. + /// Thread-safe and guarantees single initialization via Lazy. + /// + private async Task EnsureInitializedAsync() + { + await _initializationTask.Value; + } + + public async Task EnqueueAsync(EmailOutbox email, CancellationToken cancellationToken = default) + { + await EnsureInitializedAsync(); // Initialize on first call + + var json = JsonSerializer.Serialize(email); + var body = Encoding.UTF8.GetBytes(json); + + var properties = new BasicProperties + { + Persistent = true, // Message persistence + ContentType = "application/json", + Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) + }; + + await _channel!.BasicPublishAsync( + exchange: ExchangeName, + routingKey: RoutingKey, + mandatory: false, + basicProperties: properties, + body: body, + cancellationToken: cancellationToken); + } + + public async Task DequeueAsync(CancellationToken cancellationToken = default) + { + await EnsureInitializedAsync(); // Initialize on first call + + var result = await _channel!.BasicGetAsync(QueueName, false, cancellationToken); + + if (result == null) + return null; + + try + { + var json = Encoding.UTF8.GetString(result.Body.ToArray()); + var email = JsonSerializer.Deserialize(json); + + // Acknowledge message after successful deserialization + await _channel.BasicAckAsync(result.DeliveryTag, false, cancellationToken); + + return email; + } + catch + { + // Reject and requeue message on error + await _channel.BasicNackAsync(result.DeliveryTag, false, true, cancellationToken); + throw; + } + } + + public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default) + { + await EnsureInitializedAsync(); // Initialize on first call + + var queueInfo = await _channel!.QueueDeclarePassiveAsync(QueueName, cancellationToken); + return (int)queueInfo.MessageCount; + } + + /// + /// Start event-driven consumer that processes messages as they arrive + /// + public async Task StartConsumerAsync( + Func onMessageReceived, + CancellationToken cancellationToken = default) + { + await EnsureInitializedAsync(); // Initialize on first call + + var consumer = new AsyncEventingBasicConsumer(_channel!); + + consumer.ReceivedAsync += async (sender, args) => + { + try + { + var json = Encoding.UTF8.GetString(args.Body.ToArray()); + var email = JsonSerializer.Deserialize(json); + + if (email != null) + { + _logger.LogDebug("Received email message: To={To}, Subject={Subject}", email.Recipient, email.Subject); + + // Process message via callback + await onMessageReceived(email); + + // Acknowledge message after successful processing + await _channel.BasicAckAsync(args.DeliveryTag, false, cancellationToken); + _logger.LogDebug("Message acknowledged: DeliveryTag={DeliveryTag}", args.DeliveryTag); + } + else + { + _logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag); + await _channel.BasicNackAsync(args.DeliveryTag, false, false, cancellationToken); // Don't requeue invalid messages + } + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to process email message: DeliveryTag={DeliveryTag}. Moving to DLQ (NO retry).", args.DeliveryTag); + + // TODO: Error Reporting Strategy + // Option 1: Separate RabbitMQ Queue (emailprofiler.errors) + // - Create EmailErrorReport entity { EmailOutboxId, Exception, StackTrace, Timestamp, RetryAttempt } + // - Publish to error queue: await _errorQueue.EnqueueAsync(errorReport) + // - Separate worker processes error queue → Log to DB/File/External monitoring + // + // Option 2: Database Table (TBEMLP_ERROR_LOG) + // - Columns: ERROR_ID, OUTBOX_ID, ERROR_MESSAGE, STACK_TRACE, ERROR_DATE + // - Insert via IErrorLogRepository.CreateAsync(errorLog) + // + // Option 3: External Monitoring Service + // - Sentry: SentrySdk.CaptureException(ex) + // - Application Insights: _telemetryClient.TrackException(ex) + // - Elasticsearch: _elasticClient.IndexDocument(errorLog) + // + // Recommended: Option 1 (RabbitMQ Error Queue) + Option 2 (DB persistence) + // - Fast async error logging (non-blocking) + // - Persistent storage for audit + // - Real-time alerting via monitoring worker + + // NO RETRY - All failures move directly to DLQ + await _channel.BasicNackAsync(args.DeliveryTag, false, false, cancellationToken); // requeue=false → DLQ + } + }; + + // Start consuming messages (event-driven, non-blocking) + await _channel.BasicConsumeAsync( + queue: QueueName, + autoAck: false, + consumer: consumer, + cancellationToken: cancellationToken); + + _logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", QueueName); + } + + public void Dispose() + { + _channel?.CloseAsync().GetAwaiter().GetResult(); + _channel?.Dispose(); + _connection?.CloseAsync().GetAwaiter().GetResult(); + _connection?.Dispose(); + } +} diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Services/LimilabsEmailService.cs b/src/DigitalData.EmailProfiler.Infrastructure/Services/LimilabsEmailService.cs new file mode 100644 index 0000000..2c22ac2 --- /dev/null +++ b/src/DigitalData.EmailProfiler.Infrastructure/Services/LimilabsEmailService.cs @@ -0,0 +1,103 @@ +using DigitalData.EmailProfiler.Application.Common.Dtos; +using DigitalData.EmailProfiler.Application.Common.Interfaces; +using DigitalData.EmailProfiler.Domain.Exceptions; +using Limilabs.Client.SMTP; +using Limilabs.Mail; +using Limilabs.Mail.Headers; +using Microsoft.Extensions.Options; + +namespace DigitalData.EmailProfiler.Infrastructure.Services; + +/// +/// Email service using Limilabs Mail.dll for SMTP operations (send-only). +/// Commercial-grade library with superior Exchange support. +/// SMTP configuration is injected via IOptions<EmailAccountDto> from appsettings.json. +/// +public class LimilabsEmailService( + IEncryptionService encryptionService, + IOptions smtpConfig) : IEmailService +{ + private readonly EmailAccountDto _smtpAccount = smtpConfig.Value; + + public async Task SendEmailAsync(string to, string subject, string body, bool isHtml = true, CancellationToken cancellationToken = default) + { + using var smtp = new Smtp(); + + try + { + await ConnectAndAuthenticateSmtpAsync(smtp); + + var builder = new MailBuilder(); + builder.From.Add(new MailBox(_smtpAccount.Username)); + builder.To.Add(new MailBox(to)); + builder.Subject = subject; + + if (isHtml) + { + builder.Html = body; + } + else + { + builder.Text = body; + } + + var mail = builder.Create(); + + var result = smtp.SendMessage(mail); + + if (result.Status != SendMessageStatus.Success) + { + throw new InvalidOperationException($"Failed to send email. Status: {result.Status}"); + } + + smtp.Close(); + await Task.CompletedTask; // For async consistency + } + catch (Limilabs.Client.ServerException ex) + { + DisconnectSafely(smtp); + throw new AuthenticationFailedException("SMTP authentication failed. Check credentials or OAuth2 configuration.", ex); + } + catch (Exception ex) + { + DisconnectSafely(smtp); + throw new InvalidOperationException("Failed to send email via SMTP server.", ex); + } + } + + // --- Private Helper Methods --- + + private async Task ConnectAndAuthenticateSmtpAsync(Smtp smtp) + { + if (_smtpAccount.SmtpUseSsl) + { + smtp.ConnectSSL(_smtpAccount.SmtpServer, _smtpAccount.SmtpPort); + } + else + { + smtp.Connect(_smtpAccount.SmtpServer, _smtpAccount.SmtpPort); + } + + if (_smtpAccount.UseOAuth2) + { + throw new NotSupportedException("OAuth2 is not configured for this SMTP account. UseOAuth2 must be false."); + } + else + { + var password = encryptionService.Decrypt(_smtpAccount.EncryptedPassword!); + smtp.Login(_smtpAccount.Username, password); + } + + await Task.CompletedTask; // For async consistency + } + + private static void DisconnectSafely(Smtp smtp) + { + try + { + if (smtp.Connected) + smtp.Close(); + } + catch { /* Ignore disconnect errors */ } + } +}