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

refactor(infrastructure): Refactor RabbitMqEmailQueue and remove InMemoryEmailQueue, update DbContext

Browse Source
This commit is contained in:
tekh
2026-07-23 11:19:59 +02:00
parent 71e29ac3bb
commit 3be6e28477
3 changed files with 10 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/DigitalData.EmailProfiler.Infrastructure/Persistence/EmailProfilerDbContext.cs
View File

@@ -1,4 +1,3 @@
using DigitalData.EmailProfiler.Domain.Entities;
using Microsoft.EntityFrameworkCore;
namespace DigitalData.EmailProfiler.Infrastructure.Persistence;
@@ -9,13 +8,4 @@ namespace DigitalData.EmailProfiler.Infrastructure.Persistence;
/// </summary>
public class EmailProfilerDbContext(DbContextOptions<EmailProfilerDbContext> options) : DbContext(options)
{
// DbSets for all entities
public DbSet<EmailAccount> EmailAccounts { get; set; }
public DbSet<EmailProfile> EmailProfiles { get; set; }
public DbSet<EmailHistory> EmailHistories { get; set; }
public DbSet<EmailAttachment> EmailAttachments { get; set; }
public DbSet<EmailProcess> EmailProcesses { get; set; }
public DbSet<ProcessStep> ProcessSteps { get; set; }
public DbSet<IndexingStep> IndexingSteps { get; set; }
public DbSet<EmailOutbox> EmailOutbox { get; set; }
}

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

@@ -1,59 +0,0 @@
using System.Threading.Channels;
using DigitalData.EmailProfiler.Application.Common.Interfaces;
using DigitalData.EmailProfiler.Domain.Entities;
namespace DigitalData.EmailProfiler.Infrastructure.Queue;
/// <summary>
/// In-memory email queue implementation using System.Threading.Channels.
/// Thread-safe, high-performance queue for outgoing emails.
///
/// 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;
public InMemoryEmailQueue()
{
var options = new BoundedChannelOptions(1000)
{
FullMode = BoundedChannelFullMode.Wait
};
_channel = Channel.CreateBounded<EmailOutbox>(options);
}
public async Task EnqueueAsync(EmailOutbox email, CancellationToken cancellationToken = default)
{
await _channel.Writer.WriteAsync(email, cancellationToken);
}
public async Task<EmailOutbox?> DequeueAsync(CancellationToken cancellationToken = default)
{
if (await _channel.Reader.WaitToReadAsync(cancellationToken))
{
if (_channel.Reader.TryRead(out var email))
{
return email;
}
}
return null;
}
public Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
{
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.");
}
}

20
src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs
View File

@@ -1,13 +1,13 @@
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.Application.EmailSending.Commands;
using DigitalData.EmailProfiler.Infrastructure.Messaging;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using DigitalData.EmailProfiler.Application.Common.Events;
namespace DigitalData.EmailProfiler.Infrastructure.Queue;
@@ -126,11 +126,11 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
await _initializationTask.Value;
}
public async Task EnqueueAsync(EmailOutbox email, CancellationToken cancellationToken = default)
public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default)
{
await EnsureInitializedAsync(); // Initialize on first call
var json = JsonSerializer.Serialize(email);
var json = JsonSerializer.Serialize(outgoingEmailEvent);
var body = Encoding.UTF8.GetBytes(json);
var properties = new BasicProperties
@@ -149,7 +149,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
cancellationToken: cancellationToken);
}
public async Task<EmailOutbox?> DequeueAsync(CancellationToken cancellationToken = default)
public async Task<OutgoingEmailEvent?> DequeueAsync(CancellationToken cancellationToken = default)
{
await EnsureInitializedAsync(); // Initialize on first call
@@ -161,7 +161,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
try
{
var json = Encoding.UTF8.GetString(result.Body.ToArray());
var email = JsonSerializer.Deserialize<EmailOutbox>(json);
var email = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);
// Acknowledge message after successful deserialization
await _channel.BasicAckAsync(result.DeliveryTag, false, cancellationToken);
@@ -188,7 +188,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
/// Start event-driven consumer that processes messages as they arrive
/// </summary>
public async Task StartConsumerAsync(
Func<EmailOutbox, Task> onMessageReceived,
Func<OutgoingEmailEvent, Task> onMailReceived,
CancellationToken cancellationToken = default)
{
await EnsureInitializedAsync(); // Initialize on first call
@@ -200,14 +200,14 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
try
{
var json = Encoding.UTF8.GetString(args.Body.ToArray());
var email = JsonSerializer.Deserialize<EmailOutbox>(json);
var email = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);
if (email != null)
{
_logger.LogDebug("Received email message: To={To}, Subject={Subject}", email.Recipient, email.Subject);
// Process message via callback
await onMessageReceived(email);
await onMailReceived(email);
// Acknowledge message after successful processing
await _channel.BasicAckAsync(args.DeliveryTag, false, cancellationToken);
@@ -225,7 +225,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
// TODO: Error Reporting Strategy
// Option 1: Separate RabbitMQ Queue (emailprofiler.errors)
// - Create EmailErrorReport entity { EmailOutboxId, Exception, StackTrace, Timestamp, RetryAttempt }
// - Create EmailErrorReport entity { OutgoingEmailEventId, Exception, StackTrace, Timestamp, RetryAttempt }
// - Publish to error queue: await _errorQueue.EnqueueAsync(errorReport)
// - Separate worker processes error queue → Log to DB/File/External monitoring
//