Atomic

Domain Events

Create (aka Raise) and Consume (aka Publish) events https://www.youtube.com/watch?v=AHzWJ_SMqLo

Usage flow

• EntityFramework (EF) will track changes to the entity objects
• Each entity object will have a list of domain events for the given entity
• When performing ******Outbox Pattern the entities changes tracked are fetched (via ChangeTracker) and obtain all domain events
• Create outbox message for each domain event
• Save all outbox message
• EF generates sql command for all changes (original and new outbox messages changes) to ensure atomicity
• Commits transaction

Outbox Pattern

Intercept db action and modify db context to include another change in the same transaction Then consume those intercepted atomic entries separately https://www.youtube.com/watch?v=XALvnX7MPeo https://chatgpt.com/share/67da2996-db10-8001-a544-b71f1a56e105

public async Task PlaceNewOrder()
{
    var order = new Order("12345");
    
    // 1) This raises a domain event inside the AggregateRoot
    order.PlaceOrder(); 

	// 2) Entity is now tracked by ChangeTracker
    _dbContext.Orders.Add(order); 

	// 3) The interceptor will process this!
    await _dbContext.SaveChangesAsync(); 
}


// 1) domain events


public class Order : AggregateRoot
{
    public string OrderNumber { get; private set; }

    public void PlaceOrder()
    {
        // Raise a domain event when an order is placed
        RaiseDomainEvent(new OrderPlacedEvent(this.Id));
    }
}

public class AggregateRoot : Entity
{
	// multiple domain events per object
	private List<DomainEvents> _domainEvents == new();

	// add
	public RaiseDomainEvent(DomainEvent domainEvent);

	// get
	public GetDomainEvents();
	
	// clear
	public ClearDomainEvents();
}


// 2) interceptor


public class DomainEventsInterceptor : SaveChangesInterceptor
{
    public override async ValueTask<int> SavedChangesAsync(
        SaveChangesCompletedEventData eventData, 
        int result, 
        CancellationToken cancellationToken = default)
    {
        var context = eventData.Context;
        if (context == null) return result;

        // Step 1: Get all domain events from AggregateRoots
        var domainEvents = context.ChangeTracker.Entries<AggregateRoot>()
            .SelectMany(e => e.Entity.GetDomainEvents())
            .ToList();

        if (domainEvents.Any())
        {
            // Step 2: Convert domain events into Outbox Messages
            var outboxMessages = domainEvents.Select(e => new OutboxMessage(e)).ToList();

            // Step 3: Persist outbox messages
            await context.Set<OutboxMessage>().AddRangeAsync(outboxMessages, cancellationToken);

            // Step 4: Call base method to persist outbox messages (and the original db changes)
            await base.SavedChangesAsync(eventData, result, cancellationToken);

            // Step 5: Clear domain events from AggregateRoot
            foreach (var entity in context.ChangeTracker.Entries<AggregateRoot>())
            {
                entity.Entity.ClearDomainEvents();
            }
        }

        return result;
    }
}

then a background job can consume the outbox messages

benefits:

• domain events decoupled events
• outbox takes domain events and makes it atomic with db commit

drawbacks:

• if process goes down while processing events -> need Stateful