AppStd/DigitalData.MessagingService
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(infrastructure): Add Limilabs email service and RabbitMQ email queue

Browse Source 
- 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)
This commit is contained in:
tekh
2026-07-22 11:48:37 +02:00
parent dbd0d35ba3
commit 3f9bfc78a8
6 changed files with 402 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs
View File

@@ -10,4 +10,11 @@ public interface IEmailQueue
Task EnqueueAsync(EmailOutbox email, CancellationToken cancellationToken = default);
Task<EmailOutbox?> DequeueAsync(CancellationToken cancellationToken = default);
Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Start event-driven consumer that calls callback when message received
/// </summary>
/// <param name="onMessageReceived">Callback to process received message</param>
/// <param name="cancellationToken">Cancellation token</param>
Task StartConsumerAsync(Func<EmailOutbox, Task> onMessageReceived, CancellationToken cancellationToken = default);
}

27
src/DigitalData.EmailProfiler.Infrastructure/DependencyInjection.cs
View File

@@ -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<EmailProfilerDbContext>(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<IEmailService, MailKitEmailService>();
// Email Service (using Limilabs Mail.dll - Singleton for use in EmailSenderWorker)
services.AddSingleton<IEmailService, LimilabsEmailService>();
// PDF Processing Service (using DevExpress.Pdf)
services.AddScoped<IPdfProcessingService, DevExpressPdfProcessingService>();
// DMS Service (using windream COM Interop)
services.AddScoped<IDmsService, WindreamDmsService>();
// Encryption Service (using Data Protection API)
services.AddScoped<IEncryptionService, DataProtectionEncryptionService>();
// Encryption Service (using Data Protection API - Singleton, thread-safe)
services.AddSingleton<IEncryptionService, DataProtectionEncryptionService>();
// --- Email Queue ---
services.AddSingleton<IEmailQueue, InMemoryEmailQueue>();
// --- Email Queue (RabbitMQ) ---
services.AddSingleton<IEmailQueue, RabbitMqEmailQueue>();
// --- RabbitMQ Configuration ---
services.Configure<RabbitMqConfiguration>(

6
src/DigitalData.EmailProfiler.Infrastructure/DigitalData.EmailProfiler.Infrastructure.csproj
View File

@@ -27,4 +27,10 @@
<PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
</ItemGroup>
<ItemGroup>
<Reference Include="Mail">
<HintPath>M:\Bibliotheken\3rdParty\Limilabs\Mail.dll</HintPath>
</Reference>
</ItemGroup>
</Project>

13
src/DigitalData.EmailProfiler.Infrastructure/Queue/InMemoryEmailQueue.cs
View File

@@ -7,8 +7,11 @@ namespace DigitalData.EmailProfiler.Infrastructure.Queue;
/// <summary>
/// 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.
/// </summary>
[Obsolete("InMemoryEmailQueue is obsolete. Use RabbitMqEmailQueue for production deployment.")]
public class InMemoryEmailQueue : IEmailQueue
{
private readonly Channel<EmailOutbox> _channel;
@@ -45,4 +48,12 @@ public class InMemoryEmailQueue : IEmailQueue
{
return Task.FromResult(_channel.Reader.Count);
}
/// <summary>
/// NOT IMPLEMENTED - InMemoryEmailQueue does not support event-driven consumers
/// </summary>
public Task StartConsumerAsync(Func<EmailOutbox, Task> onMessageReceived, CancellationToken cancellationToken = default)
{
throw new NotSupportedException("InMemoryEmailQueue does not support StartConsumerAsync. Use RabbitMqEmailQueue for event-driven consumers.");
}
}

268
src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs Normal file
View File

@@ -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;
/// <summary>
/// RabbitMQ-based email queue implementation for outgoing emails.
/// Provides message persistence, scalability, and reliability.
/// Uses Lazy<T> initialization pattern to avoid blocking constructor.
/// </summary>
public class RabbitMqEmailQueue : IEmailQueue, IDisposable
{
private readonly ILogger<RabbitMqEmailQueue> _logger;
private readonly RabbitMqConfiguration _config;
private IConnection? _connection;
private IChannel _channel;
// Lazy<T> ensures InitAsync is called only ONCE (thread-safe)
private readonly Lazy<Task> _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<RabbitMqConfiguration> config, ILogger<RabbitMqEmailQueue> logger)
#pragma warning restore CS8618
{
_logger = logger;
_config = config.Value;
// Lazy initialization - NOT executed until first access (non-blocking constructor)
_initializationTask = new Lazy<Task>(InitAsync);
}
/// <summary>
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary>
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<string, object?>
{
{ "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);
}
/// <summary>
/// Ensure RabbitMQ is initialized before any operation.
/// Thread-safe and guarantees single initialization via Lazy<T>.
/// </summary>
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<EmailOutbox?> 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<EmailOutbox>(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<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
{
await EnsureInitializedAsync(); // Initialize on first call
var queueInfo = await _channel!.QueueDeclarePassiveAsync(QueueName, cancellationToken);
return (int)queueInfo.MessageCount;
}
/// <summary>
/// Start event-driven consumer that processes messages as they arrive
/// </summary>
public async Task StartConsumerAsync(
Func<EmailOutbox, Task> 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<EmailOutbox>(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();
}
}

103
src/DigitalData.EmailProfiler.Infrastructure/Services/LimilabsEmailService.cs Normal file
View File

@@ -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;
/// <summary>
/// Email service using Limilabs Mail.dll for SMTP operations (send-only).
/// Commercial-grade library with superior Exchange support.
/// SMTP configuration is injected via IOptions&lt;EmailAccountDto&gt; from appsettings.json.
/// </summary>
public class LimilabsEmailService(
IEncryptionService encryptionService,
IOptions<EmailAccountDto> 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 */ }
}
}