init

Femto.Modules.Media/Infrastructure/Integration/Mailman.cs

@@ -0,0 +1,78 @@
+using System.Text.Json;
+using Femto.Modules.Media.Data;
+using MediatR;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace Femto.Modules.Media.Infrastructure.Integration;
+
+internal class Mailman(Outbox outbox, MediaContext context, ILogger<Mailman> logger, IMediator mediator)
+    : BackgroundService
+{
+    protected override async Task ExecuteAsync(CancellationToken cancellationToken)
+    {
+        var timeToWait = TimeSpan.FromSeconds(1);
+
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            try
+            {
+                await this.DeliverMail(cancellationToken);
+            }
+            catch (Exception e)
+            {
+                logger.LogError(e, "Error while processing outbox");
+            }
+
+            try
+            {
+                await Task.Delay(timeToWait, cancellationToken);
+            }
+            catch (TaskCanceledException)
+            {
+                break;
+            }
+        }
+    }
+
+    private async Task DeliverMail(CancellationToken cancellationToken)
+    {
+        var messages = await outbox.GetPendingMessages(cancellationToken);
+
+        foreach (var message in messages)
+        {
+            try
+            {
+                var notificationType = OutboxMessageTypeRegistry.GetType(message.EventType);
+                if (notificationType is null)
+                {
+                    logger.LogWarning("unmapped event type {Type}. skipping.", message.EventType);
+                    continue;
+                }
+
+                var notification =
+                    JsonSerializer.Deserialize(message.Payload, notificationType) as INotification;
+
+                if (notification is null)
+                    throw new Exception("notification is null");
+
+                await mediator.Publish(notification, cancellationToken);
+
+                message.Succeed();
+            }
+            catch (Exception e)
+            {
+                logger.LogError(
+                    e,
+                    "Error processing event {EventId} for aggregate {AggregateId}",
+                    message.Id,
+                    message.AggregateId
+                );
+
+                message.Fail(e.ToString());
+            }
+
+            await context.SaveChangesAsync(cancellationToken);
+        }
+    }
+}

Femto.Modules.Media/Infrastructure/Integration/Outbox.cs

@@ -0,0 +1,34 @@
+using System.Text.Json;
+using Femto.Common.Integration;
+using Femto.Modules.Media.Data;
+using Microsoft.EntityFrameworkCore;
+
+namespace Femto.Modules.Media.Infrastructure.Integration;
+
+internal class Outbox(MediaContext context)
+{
+    public async Task AddMessage<TMessage>(Guid aggregateId, TMessage message, CancellationToken cancellationToken)
+        where TMessage : IIntegrationEvent
+    {
+        await context.Outbox.AddAsync(
+            new(
+                message.EventId,
+                aggregateId,
+                typeof(TMessage).Name,
+                JsonSerializer.Serialize(message)
+            ),
+            cancellationToken
+        );
+    }
+
+    public async Task<IEnumerable<OutboxEntry>> GetPendingMessages(CancellationToken cancellationToken)
+    {
+        var now = DateTime.UtcNow;
+
+        return await context
+            .Outbox.Where(message => message.Status == OutboxEntryStatus.Pending)
+            .Where(message => message.NextRetryAt == null || message.NextRetryAt <= now)
+            .OrderBy(message => message.CreatedAt)
+            .ToListAsync(cancellationToken);
+    }
+}

Femto.Modules.Media/Infrastructure/Integration/OutboxEntry.cs

@@ -0,0 +1,59 @@
+namespace Femto.Modules.Media.Infrastructure.Integration;
+
+internal class OutboxEntry
+{
+    private const int MaxRetries = 5;
+    
+    public Guid Id { get; private set; }
+
+    public string EventType { get; private set; } = null!;
+    public Guid AggregateId { get; private set; }
+
+    public string Payload { get; private set; } = null!;
+
+    public DateTime CreatedAt { get; private set; }
+
+    public DateTime? ProcessedAt { get; private set; }
+    public DateTime? NextRetryAt { get; private set; }
+    public int RetryCount { get; private set; } = 0;
+    public string? LastError { get; private set; }
+    public OutboxEntryStatus Status { get; private set; }
+    
+    private OutboxEntry() { }
+
+    public OutboxEntry(Guid eventId, Guid aggregateId, string eventType, string payload)
+    {
+        this.Id = eventId;
+        this.EventType = eventType;
+        this.AggregateId = aggregateId;
+        this.Payload = payload;
+        this.CreatedAt = DateTime.UtcNow;
+    }
+
+    public void Succeed()
+    {
+        this.ProcessedAt = DateTime.UtcNow;
+        this.Status = OutboxEntryStatus.Completed;
+    }
+    
+    public void Fail(string error)
+    {
+        if (this.RetryCount >= MaxRetries)
+        {
+            this.Status = OutboxEntryStatus.Failed;
+        }
+        else
+        {
+            this.LastError = error;
+            this.NextRetryAt = DateTime.UtcNow.AddSeconds(Math.Pow(2, this.RetryCount));
+            this.RetryCount++;
+        }
+    }
+}
+
+public enum OutboxEntryStatus
+{
+    Pending,
+    Completed,
+    Failed
+}

Femto.Modules.Media/Infrastructure/Integration/OutboxMessageTypeRegistry.cs

@@ -0,0 +1,35 @@
+using System.Collections.Concurrent;
+using System.Reflection;
+using Femto.Common.Attributes;
+using MediatR;
+
+namespace Femto.Modules.Media.Infrastructure.Integration;
+
+internal static class OutboxMessageTypeRegistry
+{
+    private static readonly ConcurrentDictionary<string, Type> Mapping = new();
+
+    public static void RegisterOutboxMessageTypesInAssembly(Assembly assembly)
+    {
+        var types = assembly.GetTypes();
+
+        foreach (var type in types)
+        {
+            if (!typeof(INotification).IsAssignableFrom(type) || type.IsAbstract || type.IsInterface)
+                continue;
+
+            var attribute = type.GetCustomAttribute<EventTypeAttribute>();
+            if (attribute == null)
+                continue;
+
+            var eventName = attribute.Name;
+            if (!string.IsNullOrWhiteSpace(eventName))
+            {
+                Mapping.TryAdd(eventName, type);
+            }
+        }
+    }
+    
+    public static Type? GetType(string eventName) => Mapping.GetValueOrDefault(eventName);
+
+}