Refactor email publisher integration

Replaced direct registration of `OutgoingEmailPublisher` with `AddMessagingServicePublisher()` to centralize publisher setup. Removed `OutgoingEmailPublisher.cs` and its dependencies, indicating a shift to a new implementation. Updated `DependencyInjection.cs` to use `DigitalData.MessagingService.Publisher` instead of the abstraction layer. Added a project reference to `DigitalData.MessagingService.Publisher` in the infrastructure project file.

src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs

@@ -2,7 +2,7 @@ using DigitalData.MessagingService.Application.Common.Interfaces;
 using DigitalData.MessagingService.Infrastructure.Queue;
 using DigitalData.MessagingService.Infrastructure.Services;
 using DigitalData.MessagingService.Infrastructure.Services.Background;
-using DigitalData.MessagingService.Publisher.Abstraction;
+using DigitalData.MessagingService.Publisher;
 using DigitalData.MessagingService.RabbitMQ;
 using Microsoft.AspNetCore.DataProtection;
 using Microsoft.Extensions.Configuration;
@@ -34,7 +34,7 @@ public static class DependencyInjection
 
         // --- Email Queue (RabbitMQ) ---
         services.AddSingleton<OutgoingEmailConsumer>();
-        services.AddSingleton<IOutgoingEmailPublisher, OutgoingEmailPublisher>();
+        services.AddMessagingServicePublisher();
 
         // --- RabbitMQ Configuration ---
         services.AddRabbitMqConnectionFactory(configuration);

src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj

@@ -9,6 +9,7 @@
   <ItemGroup>
     <ProjectReference Include="..\..\core\DigitalData.MessagingService.Application\DigitalData.MessagingService.Application.csproj" />
     <ProjectReference Include="..\..\core\DigitalData.MessagingService.Domain\DigitalData.MessagingService.Domain.csproj" />
+    <ProjectReference Include="..\DigitalData.MessagingService.Publisher\DigitalData.MessagingService.Publisher.csproj" />
     <ProjectReference Include="..\DigitalData.MessagingService.RabbitMQ\DigitalData.MessagingService.RabbitMQ.csproj" />
   </ItemGroup>
 

src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs

@@ -1,107 +0,0 @@
-using System.Text;
-using System.Text.Json;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
-using RabbitMQ.Client;
-using DigitalData.MessagingService.RabbitMQ;
-using DigitalData.MessagingService.Publisher.Abstraction;
-
-namespace DigitalData.MessagingService.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 sealed class OutgoingEmailPublisher : IOutgoingEmailPublisher, IAsyncDisposable
-{
-    private readonly RabbitMqConfiguration _config;
-    private readonly ILogger<OutgoingEmailPublisher> _logger;
-    private readonly RabbitMqConnectionFactory _cnnFactory;
-    private readonly Lazy<Task<IChannel>> _lazyChannel;
-
-    public OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailPublisher> logger, RabbitMqConnectionFactory cnnFactory)
-    {
-        _config = config.Value;
-        _logger = logger;
-        _cnnFactory = cnnFactory;
-        _lazyChannel = new(InitChannelAsync);
-    }
-
-    /// <summary>
-    /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
-    /// Called lazily on first use via EnsureInitializedAsync.
-    /// </summary>
-    private async Task<IChannel> InitChannelAsync()
-    {
-        var channel = await _cnnFactory.CreateChannelAsync();
-
-        // Topology declaration can use either channel; use publish channel here
-        // Declare Dead Letter Queue (DLQ) exchange
-        await channel.ExchangeDeclareAsync(exchange: _config.DlqExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: _cnnFactory.CancellationToken);
-
-        // Declare Dead Letter Queue (DLQ)
-        await channel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: _cnnFactory.CancellationToken);
-
-        // Bind DLQ to DLQ exchange
-        await channel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: _cnnFactory.CancellationToken);
-
-        // Declare main exchange (Direct type for routing)
-        await channel.ExchangeDeclareAsync(exchange: _config.ExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: _cnnFactory.CancellationToken);
-
-        // Declare main queue (durable for persistence) with DLQ arguments
-        var queueArgs = new Dictionary<string, object?>
-        {
-            { "x-dead-letter-exchange", _config.DlqExchangeName },
-            { "x-dead-letter-routing-key", _config.DlqRoutingKey }
-        };
-
-        await channel.QueueDeclareAsync(queue: _config.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs, cancellationToken: _cnnFactory.CancellationToken);
-
-        // Bind main queue to exchange with routing key
-        await channel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: _cnnFactory.CancellationToken);
-
-        _logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName);
-
-        return channel;
-    }
-    
-    public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default)
-    {
-        var json = JsonSerializer.Serialize(outgoingEmailEvent);
-        var body = Encoding.UTF8.GetBytes(json);
-
-        var properties = new BasicProperties
-        {
-            Persistent = true, // Message persistence
-            ContentType = "application/json",
-            Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
-        };
-
-        var channel = await _lazyChannel.Value;
-
-        await channel.BasicPublishAsync(
-            exchange: _config.ExchangeName,
-            routingKey: _config.RoutingKey,
-            mandatory: false,
-            basicProperties: properties,
-            body: body,
-            cancellationToken: cancellationToken);
-    }
-
-    public async Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
-    {
-        var channel = await _lazyChannel.Value;
-        var queueInfo = await channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);
-        return (int)queueInfo.MessageCount;
-    }
-
-    public async ValueTask DisposeAsync()
-    {
-        if (await _lazyChannel.Value is IChannel channel)
-        {
-            await channel.CloseAsync();
-            await channel.DisposeAsync();
-        }
-    }
-}