Wie DDD uns geholfen hat, neue Revisionen in Pizzerien zu erstellen

public sealed class AddMeasurementCommand
{
    // ctor

    public double? Value { get; }
    public int Version { get; }
    public UUId MaterialTypeId { get; }
    public UUId MeasurementId { get; }
    public UnitOfMeasure UnitOfMeasure { get; }
    public UUId InventoryZoneId { get; }
}

public class MeasurementEvent : IPublicInventoryEvent
{
    public UUId MaterialTypeId { get; set; }
    public double? Value { get; set; }
	
    public UUId MeasurementId { get; set; }
    public int MeasurementVersion { get; set; }
    public UUId AggregateId { get; set; }
    public int Version { get; set; }
    public UnitOfMeasure UnitOfMeasure { get; set; }
    public UUId InventoryZoneId { get; set; }
}

public sealed class Inventory : IEquatable<Inventory>
{
    private readonly List<IInventoryEvent> _changes = new List<IInventoryEvent>();

    private readonly List<InventoryMeasurement> _inventoryMeasurements = new List<InventoryMeasurement>();

    internal Inventory(UUId id, int version, UUId unitId, UUId inventoryTemplateId,
        UUId startedBy, InventoryState state, DateTime startedAtUtc, DateTime? finishedAtUtc)
	
        : this(id)
    {
        Version = version;
        UnitId = unitId;
        InventoryTemplateId = inventoryTemplateId;
        StartedBy = startedBy;
        State = state;
        StartedAtUtc = startedAtUtc;
        FinishedAtUtc = finishedAtUtc;
	
    }

    private Inventory(UUId id)
    {
        Id = id;
        Version = 0;
        State = InventoryState.Unknown;
    }
	
    public UUId Id { get; private set; }
    public int Version { get; private set; }
    public UUId UnitId { get; private set; }
    public UUId InventoryTemplateId { get; private set; }
    public UUId StartedBy { get; private set; }
    public InventoryState State { get; private set; }
    public DateTime StartedAtUtc { get; private set; }
    public DateTime? FinishedAtUtc { get; private set; }
    public ReadOnlyCollection<IInventoryEvent> Changes => _changes.AsReadOnly();
	
    public ReadOnlyCollection<InventoryMeasurement> Measurements => _inventoryMeasurements.AsReadOnly();

    public static Inventory Restore(UUId inventoryId, IInventoryEvent[] events)
    {
        var inventory = new Inventory(inventoryId);
        inventory.ReplayEvents(events);
        return inventory;
    }

    public static Inventory Restore(UUId id, int version, UUId unitId, UUId inventoryTemplateId,
        UUId startedBy, InventoryState state, DateTime startedAtUtc, DateTime? finishedAtUtc,
        InventoryMeasurement[] measurements)
    {
        var inventory = new Inventory(id, version, unitId, inventoryTemplateId,
            startedBy, state, startedAtUtc, finishedAtUtc);

        inventory._inventoryMeasurements.AddRange(measurements);

        return inventory;
    }

    public static Inventory Create(UUId inventoryId)
    {
        if (inventoryId == null)
        {
            throw new ArgumentNullException(nameof(inventoryId));
        }

        return new Inventory(inventoryId);
    }

    public void ReplayEvents(params IInventoryEvent[] events)
    {
        if (events == null)
        {
            throw new ArgumentNullException(nameof(events));
        }

        foreach (var @event in events.OrderBy(e => e.Version))
        {
            Mutate(@event);
        }
    }

    public void AddMeasurement(AddMeasurementCommand command)
    {
        if (command == null)
        {
            throw new ArgumentNullException(nameof(command));
        }

        Apply(new MeasurementEvent
        {
            AggregateId = Id,
            Version = Version + 1,
            UnitId = UnitId,
            Value = command.Value,
            MeasurementVersion = command.Version,
            MaterialTypeId = command.MaterialTypeId,
            MeasurementId = command.MeasurementId,
            UnitOfMeasure = command.UnitOfMeasure,
            InventoryZoneId = command.InventoryZoneId
        });
    }

    private void Apply(IInventoryEvent @event)
    {
        Mutate(@event);
        _changes.Add(@event);
    }

    private void Mutate(IInventoryEvent @event)
    {
        When((dynamic) @event);
        Version = @event.Version;
    }

    private void When(MeasurementEvent e)
    {
        var existMeasurement = _inventoryMeasurements.SingleOrDefault(x => x.MeasurementId == e.MeasurementId);
        if (existMeasurement is null)
    {
        _inventoryMeasurements.Add(new InventoryMeasurement
        {
            Value = e.Value,
            MeasurementId = e.MeasurementId,
            MeasurementVersion = e.MeasurementVersion,
            PreviousValue = e.PreviousValue,
            MaterialTypeId = e.MaterialTypeId,
            UserId = e.By,
            UnitOfMeasure = e.UnitOfMeasure,
            InventoryZoneId = e.InventoryZoneId
        });
    }
    else
    {
        if (!existMeasurement.Value.HasValue)
        {
            throw new InventoryInvalidStateException("Change removed measurement");
        }

        if (existMeasurement.MeasurementVersion == e.MeasurementVersion - 1)
        {
            existMeasurement.Value = e.Value;
            existMeasurement.MeasurementVersion = e.MeasurementVersion;
            existMeasurement.UnitOfMeasure = e.UnitOfMeasure;
            existMeasurement.InventoryZoneId = e.InventoryZoneId;
        }
        else if (existMeasurement.MeasurementVersion < e.MeasurementVersion)
        {
            throw new MeasurementConcurrencyException(Id, e.MeasurementId, e.Value);
        }
        else if (existMeasurement.MeasurementVersion == e.MeasurementVersion &&
            existMeasurement.Value != e.Value)
        {
            throw new MeasurementConcurrencyException(Id, e.MeasurementId, e.Value);
        }
        else
        {
            throw new NotChangeException();
        }
    }
}

// Equals
// GetHashCode
}

public class InventoryMeasurement
{
    public UUId MeasurementId { get; set; }
    public UUId MaterialTypeId { get; set; }
    public UUId UserId { get; set; }
    public double? Value { get; set; }

    public int MeasurementVersion { get; set; }

    public UnitOfMeasure UnitOfMeasure { get; set; }

    public UUId InventoryZoneId { get; set; }
}

[Fact]
public void WhenAddMeasurementAfterStartInventory_ThenInventoryHaveOneMeasurement()
{
    var inventoryId = UUId.NewUUId();
    var inventory = Domain.Inventories.Entities.Inventory.Create(inventoryId);
    var unitId = UUId.NewUUId();
    inventory.StartInventory(Create.StartInventoryCommand()
        .WithUnitId(unitId)
        .Please());

    var materialTypeId = UUId.NewUUId();
    var measurementId = UUId.NewUUId();
    var measurementVersion = 1;
    var value = 500;
    var cmd = Create.AddMeasurementCommand()
        .WithMaterialTypeId(materialTypeId)
        .WithMeasurement(measurementId, measurementVersion)
        .WithValue(value)
        .Please();
    inventory.AddMeasurement(cmd);

    inventory.Measurements.Should().BeEquivalentTo(new InventoryMeasurement
    {
        MaterialTypeId = materialTypeId,
        MeasurementId = measurementId,
        MeasurementVersion = measurementVersion,
        Value = value,
        UnitOfMeasure = UnitOfMeasure.Quantity
    });
}

public class AddMeasurementChangeHandler 
    : IRequestHandler<AddMeasurementChangeRequest, AddMeasurementChangeResponse>
{
    // dependencies
    // ctor

    public async Task<AddMeasurementChangeResponse> Handle(
        AddMeasurementChangeRequest request,
        CancellationToken ct)
    {
        var inventory =
            await _inventoryRepository.GetAsync(request.AddMeasurementChange.InventoryId, ct);
        if (inventory == null)
        {
            throw new NotFoundException($"Inventory {request.AddMeasurementChange.InventoryId} is not found");
        }

        var user = await _usersRepository.GetAsync(request.UserId, ct);
        if (user == null)
        {
            throw new SecurityException();
        }

        var hasPermissions =
        await _authPermissionService.HasPermissionsAsync(request.CountryId, request.Token, inventory.UnitId, ct);
        if (!hasPermissions)
        {
            throw new SecurityException();
        }

        var unit = await _unitRepository.GetAsync(inventory.UnitId, ct);
        if (unit == null)
        {
            throw new InvalidRequestDataException($"Unit {inventory.UnitId} is not found");
        }

        var unitOfMeasure =

Enum.Parse<UnitOfMeasure>(request.AddMeasurementChange.MaterialTypeUnitOfMeasure);


        var addMeasurementCommand = new AddMeasurementCommand(	
            request.AddMeasurementChange.Value,
            request.AddMeasurementChange.Version,
            request.AddMeasurementChange.MaterialTypeId,
            request.AddMeasurementChange.Id,
            unitOfMeasure,
            request.AddMeasurementChange.InventoryZoneId);

        inventory.AddMeasurement(addMeasurementCommand);

        await HandleAsync(inventory, ct);

        return new AddMeasurementChangeResponse(request.AddMeasurementChange.Id, user.Id, user.GetName());
    }

    private async Task HandleAsync(Domain.Inventories.Entities.Inventory inventory, CancellationToken ct)
    {
            await _inventoryRepository.AppendEventsAsync(inventory.Changes, ct);
 
            try
            {
                await _localQueueDataService.Publish(inventory.Changes, ct);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "error occured while handling action");
            }
    }
}

internal sealed class InventoryRepository : IInventoryRepository
{
    // dependencies
    // ctor

    static InventoryRepository()
    {
        EventTypes = typeof(IEvent)
            .Assembly.GetTypes().Where(x => typeof(IEvent).IsAssignableFrom(x))
            .ToDictionary(t => t.FullName, x => x);
    }

    public async Task AppendAsync(IReadOnlyCollection<IEvent> events, CancellationToken ct)
    {
        using (var session = await _dbSessionFactory.OpenAsync())
        {
            if (events.Count == 0) return;

            try
            {
                foreach (var @event in events)
                {
                    await session.ExecuteAsync(Sql.AppendEvent,
                        new
                        {
                            @event.AggregateId,
                            @event.Version,
                            @event.UnitId,
                            Type = @event.GetType().FullName,
                            Data = JsonConvert.SerializeObject(@event),
                            CreatedDateTimeUtc = DateTime.UtcNow
                        }, cancellationToken: ct);
                }
            }
            catch (MySqlException e)
                when (e.Number == (int) MySqlErrorCode.DuplicateKeyEntry)
            {
                throw new OptimisticConcurrencyException(events.First().AggregateId, "");
            }
        }
    }

    public async Task<Domain.Models.Inventory> GetInventoryAsync(
        UUId inventoryId,
        CancellationToken ct)
    {
        var events = await GetEventsAsync(inventoryId, 0, ct);

        if (events.Any()) return Domain.Models.Inventory.Restore(inventoryId, events);

        return null;
    }
    
    private async Task<IEvent[]> GetEventsAsync(
        UUId id,
        int snapshotVersion,
        CancellationToken ct)
    {
        using (var session = await _dbSessionFactory.OpenAsync())
    {
            var snapshot = await GetInventorySnapshotAsync(session, inventoryId, ct);
            var version = snapshot?.Version ?? 0;
        
            var events = await GetEventsAsync(session, inventoryId, version, ct);
            if (snapshot != null)
            {
                snapshot.ReplayEvents(events);
                return snapshot;
            }

            if (events.Any())
            {
                return Domain.Inventories.Entities.Inventory.Restore(inventoryId, events);
            }

            return null;
        }
    }

    private async Task<Inventory> GetInventorySnapshotAsync(
        IDbSession session,
        UUId id,
        CancellationToken ct)
    {
        var record =
            await session.QueryFirstOrDefaultAsync<InventoryRecord>(Sql.GetSnapshot, new {AggregateId = id},
                cancellationToken: ct);
        return record == null ? null : Map(record);
    }

    private async Task<IInventoryEvent[]> GetEventsAsync(
        IDbSession session,
        UUId id,
        int snapshotVersion,
        CancellationToken ct)
    {
        var rows = await session.QueryAsync<EventRecord>(Sql.GetEvents,
            new
            {
                AggregateId = id,
                Version = snapshotVersion
            }, cancellationToken: ct);
        return rows.Select(Map).ToArray();
    }

    private static IEvent Map(EventRecord e)
    {
        var type = EventTypes[e.Type];
        return (IEvent) JsonConvert.DeserializeObject(e.Data, type);
    }
}

internal class EventRecord
{
    public string Type { get; set; }
    public string Data { get; set; }
}

namespace Dodo.DataCatalog.Contracts.Products.v1
{
    public class MaterialType
    {
        public UUId Id { get; set; }
        public int Version { get; set; }
        public int CountryId { get; set; }
        public UUId DepartmentId { get; set; }

        public string Name { get; set; }
        public MaterialCategory Category { get; set; }
        public UnitOfMeasure BasicUnitOfMeasure { get; set; }
        public bool IsRemoved { get; set; }
    }

    public enum UnitOfMeasure
    {
        Quantity = 1,
        Gram = 5,
        Milliliter = 7,
        Meter = 8,
    }

    public enum MaterialCategory
    {
        Ingredient = 1,
        SemiFinishedProduct = 2,
        FinishedProduct = 3,
        Inventory = 4,
        Packaging = 5,
        Consumables = 6
    }
}

public class MaterialTypeConsumer : IConsumer<Dodo.DataCatalog.Contracts.Products.v1.MaterialType>
{
    private readonly IMaterialTypeRepository _materialTypeRepository;

    public MaterialTypeConsumer(IMaterialTypeRepository materialTypeRepository)
    {
         _materialTypeRepository = materialTypeRepository;
    }
 
    public async Task Consume(ConsumeContext<Dodo.DataCatalog.Contracts.Products.v1.MaterialType> context)
    {
        var materialType = new AddMaterialType(context.Message.Id,
            context.Message.Name,
            (int)context.Message.Category,
            (int)context.Message.BasicUnitOfMeasure,
            context.Message.CountryId,
            context.Message.DepartmentId,
            context.Message.IsRemoved,
            context.Message.Version);
    
        await _materialTypeRepository.SaveAsync(materialType, context.CancellationToken);
    }
}

internal sealed class BusDataService : IBusDataService
{
    private readonly IPublisherControl _publisherControl;
    private readonly IPublicQueueRepository _repository;
    private readonly EventMapper _eventMapper;

    public BusDataService(
        IPublicQueueRepository repository,
        IPublisherControl publisherControl,
        EventMapper eventMapper)
    {
        _repository = repository;
        _publisherControl = publisherControl;
        _eventMapper = eventMapper;
    }

    public async Task ConsumePublicQueueAsync(int batchEventSize, CancellationToken cancellationToken)
    {
        var events = await _repository.GetAsync(batchEventSize, cancellationToken);
        await Publish(events, cancellationToken);
    }

    public async Task Publish(IEnumerable<IPublicInventoryEvent> events, CancellationToken ct)
    {
        foreach (var @event in events)
        {
            var publicQueueEvent = _eventMapper.Map((dynamic) @event);
            await _publisherControl.Publish(publicQueueEvent, ct);
            await _repository.DeleteAsync(@event, ct);
       }
    }
}