Add Outbox and Internal Command

.editorconfig

@@ -269,7 +269,7 @@ dotnet_diagnostic.S3903.severity = None
 
 #
 # MA0004: Use Task.ConfigureAwait(false)
-dotnet_diagnostic.MA0004.severity = Suggestion
+dotnet_diagnostic.MA0004.severity = None
 
 # MA0049: Type name should not match containing namespace
 dotnet_diagnostic.MA0049.severity = Suggestion

src/BuildingBlocks/Contracts/EventBus.Messages/FlighContracts.cs

@@ -1,4 +1,4 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
 namespace BuildingBlocks.Contracts.EventBus.Messages;
 

src/BuildingBlocks/Contracts/EventBus.Messages/IdentityContracts.cs

@@ -1,4 +1,4 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
 namespace BuildingBlocks.Contracts.EventBus.Messages;
 

src/BuildingBlocks/Contracts/EventBus.Messages/ReservationContracts.cs

@@ -1,4 +1,4 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
 namespace BuildingBlocks.Contracts.EventBus.Messages;
 

src/BuildingBlocks/Core/CQRS/ICommand.cs

@@ -0,0 +1,12 @@
+using MediatR;
+
+namespace BuildingBlocks.Core.CQRS;
+
+public interface ICommand : ICommand<Unit>
+{
+}
+
+public interface ICommand<out T> : IRequest<T>
+    where T : notnull
+{
+}

src/BuildingBlocks/Core/CQRS/ICommandHandler.cs

@@ -0,0 +1,14 @@
+using MediatR;
+
+namespace BuildingBlocks.Core.CQRS;
+
+public interface ICommandHandler<in TCommand> : ICommandHandler<TCommand, Unit>
+    where TCommand : ICommand<Unit>
+{
+}
+
+public interface ICommandHandler<in TCommand, TResponse> : IRequestHandler<TCommand, TResponse>
+    where TCommand : ICommand<TResponse>
+    where TResponse : notnull
+{
+}

src/BuildingBlocks/Core/CQRS/IQuery.cs

@@ -0,0 +1,8 @@
+using MediatR;
+
+namespace BuildingBlocks.Core.CQRS;
+
+public interface IQuery<out T> : IRequest<T>
+    where T : notnull
+{
+}

src/BuildingBlocks/Core/CQRS/IQueryHandler.cs

@@ -0,0 +1,9 @@
+using MediatR;
+
+namespace BuildingBlocks.Core.CQRS;
+
+public interface IQueryHandler<in TQuery, TResponse> : IRequestHandler<TQuery, TResponse>
+    where TQuery : IQuery<TResponse>
+    where TResponse : notnull
+{
+}

src/BuildingBlocks/Domain/Event/EventType.cs → src/BuildingBlocks/Core/Event/EventType.cs

@@ -1,4 +1,4 @@
-namespace BuildingBlocks.Domain.Event;
+namespace BuildingBlocks.Core.Event;
 
 [Flags]
 public enum EventType

src/BuildingBlocks/Core/Event/IDomainEvent.cs

@@ -0,0 +1,6 @@
+namespace BuildingBlocks.Core.Event;
+
+public interface IDomainEvent : IEvent
+{
+
+}

src/BuildingBlocks/Domain/Event/IEvent.cs → src/BuildingBlocks/Core/Event/IEvent.cs

@@ -1,7 +1,6 @@
-using MassTransit;
 using MediatR;
 
-namespace BuildingBlocks.Domain.Event;
+namespace BuildingBlocks.Core.Event;
 
 public interface IEvent : INotification
 {

src/BuildingBlocks/Domain/Event/IHaveIntegrationEvent.cs → src/BuildingBlocks/Core/Event/IHaveIntegrationEvent.cs

@@ -1,4 +1,4 @@
-namespace BuildingBlocks.Domain.Event;
+namespace BuildingBlocks.Core.Event;
 
 public interface IHaveIntegrationEvent
 {

src/BuildingBlocks/Domain/Event/IIntegrationEvent.cs → src/BuildingBlocks/Core/Event/IIntegrationEvent.cs

@@ -1,7 +1,6 @@
 using MassTransit;
-using MassTransit.Topology;
 
-namespace BuildingBlocks.Domain.Event;
+namespace BuildingBlocks.Core.Event;
 
 [ExcludeFromTopology]
 public interface IIntegrationEvent : IEvent

src/BuildingBlocks/Core/Event/IInternalCommand.cs

@@ -0,0 +1,10 @@
+using BuildingBlocks.Core.CQRS;
+
+namespace BuildingBlocks.Core.Event;
+
+public interface IInternalCommand : ICommand
+{
+    long Id { get; }
+    DateTime OccurredOn { get; }
+    string Type { get; }
+}

src/BuildingBlocks/Core/Event/InternalCommand.cs

@@ -0,0 +1,13 @@
+using BuildingBlocks.IdsGenerator;
+using BuildingBlocks.Utils;
+
+namespace BuildingBlocks.Core.Event;
+
+public class InternalCommand : IInternalCommand
+{
+    public long Id { get; set; } = SnowFlakIdGenerator.NewId();
+
+    public DateTime OccurredOn => DateTime.Now;
+
+    public string Type { get => TypeProvider.GetTypeName(GetType()); }
+}

src/BuildingBlocks/Core/Event/MessageEnvelope.cs

@@ -0,0 +1,26 @@
+using Google.Protobuf;
+
+namespace BuildingBlocks.Core.Event;
+
+public class MessageEnvelope
+{
+    public MessageEnvelope(object? message, IDictionary<string, object?>? headers = null)
+    {
+        Message = message;
+        Headers = headers ?? new Dictionary<string, object?>();
+    }
+
+    public object? Message { get; init; }
+    public IDictionary<string, object?> Headers { get; init; }
+}
+
+public class MessageEnvelope<TMessage> : MessageEnvelope
+    where TMessage : class, IMessage
+{
+    public MessageEnvelope(TMessage message, IDictionary<string, object?> header) : base(message, header)
+    {
+        Message = message;
+    }
+
+    public new TMessage? Message { get; }
+}

src/BuildingBlocks/Domain/BusPublisher.cs → src/BuildingBlocks/Core/EventDispatcher.cs

@@ -1,88 +1,71 @@
 using System.Security.Claims;
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
+using BuildingBlocks.MessageProcessor;
 using BuildingBlocks.Web;
-using MassTransit;
 using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using MessageEnvelope = BuildingBlocks.Core.Event.MessageEnvelope;
 
-namespace BuildingBlocks.Domain;
+namespace BuildingBlocks.Core;
 
-public sealed class BusPublisher : IBusPublisher
+public sealed class EventDispatcher : IEventDispatcher
 {
     private readonly IEventMapper _eventMapper;
-    private readonly ILogger<BusPublisher> _logger;
-    private readonly IPublishEndpoint _publishEndpoint;
+    private readonly ILogger<EventDispatcher> _logger;
+    private readonly IPersistMessageProcessor _persistMessageProcessor;
     private readonly IHttpContextAccessor _httpContextAccessor;
     private readonly IServiceScopeFactory _serviceScopeFactory;
 
-    public BusPublisher(IServiceScopeFactory serviceScopeFactory,
+    public EventDispatcher(IServiceScopeFactory serviceScopeFactory,
         IEventMapper eventMapper,
-        ILogger<BusPublisher> logger,
-        IPublishEndpoint publishEndpoint,
+        ILogger<EventDispatcher> logger,
+        IPersistMessageProcessor persistMessageProcessor,
         IHttpContextAccessor httpContextAccessor)
     {
         _serviceScopeFactory = serviceScopeFactory;
         _eventMapper = eventMapper;
         _logger = logger;
-        _publishEndpoint = publishEndpoint;
+        _persistMessageProcessor = persistMessageProcessor;
         _httpContextAccessor = httpContextAccessor;
     }
 
     public async Task SendAsync(IDomainEvent domainEvent,
-        CancellationToken cancellationToken = default) => await SendAsync(new[] { domainEvent }, cancellationToken);
+        CancellationToken cancellationToken = default) => await SendAsync(new[] {domainEvent}, cancellationToken);
 
     public async Task SendAsync(IReadOnlyList<IDomainEvent> domainEvents, CancellationToken cancellationToken = default)
     {
         if (domainEvents is null) return;
 
-        _logger.LogTrace("Processing integration events start...");
-
         var integrationEvents = await MapDomainEventToIntegrationEventAsync(domainEvents).ConfigureAwait(false);
 
-        if (!integrationEvents.Any()) return;
+        if (integrationEvents.Count == 0) return;
 
-        await PublishMessageToBroker(integrationEvents, cancellationToken);
-
-        _logger.LogTrace("Processing integration events done...");
+        foreach (var integrationEvent in integrationEvents)
+        {
+            await _persistMessageProcessor.PublishMessageAsync(new MessageEnvelope(integrationEvent, SetHeaders()),
+                cancellationToken);
+        }
     }
 
 
-
     public async Task SendAsync(IIntegrationEvent integrationEvent,
-        CancellationToken cancellationToken = default) => await SendAsync(new[] { integrationEvent }, cancellationToken);
+        CancellationToken cancellationToken = default) => await SendAsync(new[] {integrationEvent}, cancellationToken);
 
-    public async Task SendAsync(IReadOnlyList<IIntegrationEvent> integrationEvents, CancellationToken cancellationToken = default)
+    public async Task SendAsync(IReadOnlyList<IIntegrationEvent> integrationEvents,
+        CancellationToken cancellationToken = default)
     {
         if (integrationEvents is null) return;
 
-        _logger.LogTrace("Processing integration events start...");
-
-        await PublishMessageToBroker(integrationEvents, cancellationToken);
-
-        _logger.LogTrace("Processing integration events done...");
-    }
-
-    private async Task PublishMessageToBroker(IReadOnlyList<IIntegrationEvent> integrationEvents, CancellationToken cancellationToken)
-    {
-        foreach (var integrationEvent in integrationEvents)
-        {
-            await _publishEndpoint.Publish((object) integrationEvent, context =>
-            {
-                context.CorrelationId = _httpContextAccessor?.HttpContext?.GetCorrelationId();
-                context.Headers.Set("UserId",
-                    _httpContextAccessor?.HttpContext?.User?.FindFirstValue(ClaimTypes.NameIdentifier));
-                context.Headers.Set("UserName",
-                    _httpContextAccessor?.HttpContext?.User?.FindFirstValue(ClaimTypes.Name));
-            }, cancellationToken);
-
-            _logger.LogTrace("Publish a message with ID: {Id}", integrationEvent?.EventId);
-        }
+        await _persistMessageProcessor.PublishMessageAsync(new MessageEnvelope(integrationEvents, SetHeaders()),
+            cancellationToken);
     }
 
     private Task<IReadOnlyList<IIntegrationEvent>> MapDomainEventToIntegrationEventAsync(
         IReadOnlyList<IDomainEvent> events)
     {
+        _logger.LogTrace("Processing integration events start...");
+
         var wrappedIntegrationEvents = GetWrappedIntegrationEvents(events.ToList())?.ToList();
         if (wrappedIntegrationEvents?.Count > 0)
             return Task.FromResult<IReadOnlyList<IIntegrationEvent>>(wrappedIntegrationEvents);
@@ -101,6 +84,8 @@ private Task<IReadOnlyList<IIntegrationEvent>> MapDomainEventToIntegrationEventA
             integrationEvents.Add(integrationEvent);
         }
 
+        _logger.LogTrace("Processing integration events done...");
+
         return Task.FromResult<IReadOnlyList<IIntegrationEvent>>(integrationEvents);
     }
 
@@ -118,4 +103,14 @@ private IEnumerable<IIntegrationEvent> GetWrappedIntegrationEvents(IReadOnlyList
             yield return domainNotificationEvent;
         }
     }
+
+    private IDictionary<string, object> SetHeaders()
+    {
+        var headers = new Dictionary<string, object>();
+        headers.Add("CorrelationId", _httpContextAccessor?.HttpContext?.GetCorrelationId());
+        headers.Add("UserId", _httpContextAccessor?.HttpContext?.User?.FindFirstValue(ClaimTypes.NameIdentifier));
+        headers.Add("UserName", _httpContextAccessor?.HttpContext?.User?.FindFirstValue(ClaimTypes.Name));
+
+        return headers;
+    }
 }

src/BuildingBlocks/Domain/IBusPublisher.cs → src/BuildingBlocks/Core/IEventDispatcher.cs

@@ -1,8 +1,8 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
-namespace BuildingBlocks.Domain;
+namespace BuildingBlocks.Core;
 
-public interface IBusPublisher
+public interface IEventDispatcher
 {
     public Task SendAsync(IReadOnlyList<IDomainEvent> domainEvents, CancellationToken cancellationToken = default);
     public Task SendAsync(IDomainEvent domainEvent, CancellationToken cancellationToken = default);

src/BuildingBlocks/Domain/IEventMapper.cs → src/BuildingBlocks/Core/IEventMapper.cs

@@ -1,6 +1,6 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
-namespace BuildingBlocks.Domain;
+namespace BuildingBlocks.Core;
 
 public interface IEventMapper
 {

src/BuildingBlocks/Domain/IntegrationEventWrapper.cs → src/BuildingBlocks/Core/IntegrationEventWrapper.cs

@@ -1,6 +1,6 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
-namespace BuildingBlocks.Domain;
+namespace BuildingBlocks.Core;
 
 public record IntegrationEventWrapper<TDomainEventType>(TDomainEventType DomainEvent) : IIntegrationEvent
     where TDomainEventType : IDomainEvent;

src/BuildingBlocks/Domain/Model/Aggregate.cs → src/BuildingBlocks/Core/Model/Aggregate.cs

@@ -1,6 +1,6 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
-namespace BuildingBlocks.Domain.Model
+namespace BuildingBlocks.Core.Model
 {
     public abstract class Aggregate : Aggregate<long>
     {

src/BuildingBlocks/Domain/Model/Entity.cs → src/BuildingBlocks/Core/Model/Entity.cs

@@ -1,4 +1,4 @@
-namespace BuildingBlocks.Domain.Model;
+namespace BuildingBlocks.Core.Model;
 
 public abstract class Entity : IEntity
 {

src/BuildingBlocks/Domain/Model/IAggregate.cs → src/BuildingBlocks/Core/Model/IAggregate.cs

@@ -1,6 +1,6 @@
-using BuildingBlocks.Domain.Event;
+using BuildingBlocks.Core.Event;
 
-namespace BuildingBlocks.Domain.Model;
+namespace BuildingBlocks.Core.Model;
 
 public interface IAggregate : IEntity
 {

src/BuildingBlocks/Domain/Model/IEntity.cs → src/BuildingBlocks/Core/Model/IEntity.cs

@@ -1,4 +1,4 @@
-namespace BuildingBlocks.Domain.Model;
+namespace BuildingBlocks.Core.Model;
 
 public interface IEntity
 {