Remove EventSourcing code
DELETE
WalletWasabi.Tests/UnitTests/EventSourcing/InMemoryEventRepositoryTests.cs
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
-using WalletWasabi.EventSourcing;
-using WalletWasabi.Exceptions;
-using WalletWasabi.Interfaces.EventSourcing;
-using WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain;
-using Xunit;
-using Xunit.Abstractions;
-
-namespace WalletWasabi.Tests.UnitTests.EventSourcing;
-
-public class InMemoryEventRepositoryTests : IDisposable
-{
- private const string AggregateType = nameof(InMemoryEventRepositoryTests);
-
- /// <summary>Aggregate ID number one.</summary>
- private const string Id1 = "MY_ID_1";
-
- /// <summary>Aggregate ID number two.</summary>
- private const string Id2 = "MY_ID_2";
-
- private static readonly TimeSpan SemaphoreWaitTimeout = TimeSpan.FromSeconds(20);
-
- public InMemoryEventRepositoryTests(ITestOutputHelper output)
- {
- TestEventRepository = new TestInMemoryEventRepository(output);
- EventRepository = TestEventRepository;
- }
-
- private IEventRepository EventRepository { get; init; }
- private TestInMemoryEventRepository TestEventRepository { get; init; }
-
- [Fact]
- public async Task AppendEvents_Zero_Async()
- {
- WrappedEvent[] noEvents = Array.Empty<WrappedEvent>();
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, noEvents);
-
- IReadOnlyList<WrappedEvent> actualEvents = await EventRepository.GetEventsAsync(AggregateType, Id1);
- Assert.Empty(actualEvents);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Empty(actualAggregateIds);
- }
-
- [Fact]
- public async Task AppendEvents_One_Async()
- {
- WrappedEvent[] events = new[]
- {
- new WrappedEvent(SequenceId: 1),
- };
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events);
-
- IReadOnlyList<WrappedEvent> actualEvents = await EventRepository.GetEventsAsync(AggregateType, Id1);
- Assert.Equal(events, actualEvents);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { Id1 }, actualAggregateIds);
- }
-
- [Fact]
- public async Task AppendEvents_Two_Async()
- {
- WrappedEvent[] events = new[]
- {
- new WrappedEvent(SequenceId: 1),
- new WrappedEvent(SequenceId: 2),
- };
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events);
-
- IReadOnlyList<WrappedEvent> actualEvents = await EventRepository.GetEventsAsync(AggregateType, Id1);
- Assert.Equal(events, actualEvents);
-
- IReadOnlyList<string> aggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { Id1 }, aggregateIds);
- }
-
- [Fact]
- public async Task AppendEvents_NegativeSequenceId_Async()
- {
- WrappedEvent[] events = new[]
- {
- new WrappedEvent(SequenceId: -1)
- };
-
- async Task ActionAsync()
- {
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events).ConfigureAwait(false);
- }
-
- ArgumentException ex = await Assert.ThrowsAsync<ArgumentException>(ActionAsync);
- Assert.Contains("First event sequenceId is not a positive number.", ex.Message);
- }
-
- [Fact]
- public async Task AppendEvents_SkippedSequenceId_Async()
- {
- WrappedEvent[] events = new[]
- {
- new WrappedEvent(2)
- };
-
- async Task ActionAsync()
- {
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events).ConfigureAwait(false);
- }
-
- ArgumentException ex = await Assert.ThrowsAsync<ArgumentException>(ActionAsync);
- Assert.Contains("Invalid firstSequenceId (gap in sequence IDs) expected: '1' given: '2'", ex.Message);
- }
-
- [Fact]
- public async Task AppendEvents_OptimisticConcurrency_Async()
- {
- WrappedEvent[] events = new[]
- {
- new WrappedEvent(1)
- };
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events);
-
- async Task ActionAsync()
- {
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events).ConfigureAwait(false);
- }
-
- OptimisticConcurrencyException ex = await Assert.ThrowsAsync<OptimisticConcurrencyException>(ActionAsync);
- Assert.Contains("Conflict", ex.Message);
- }
-
- [Fact]
- public async Task AppendEventsAsync_Interleaving_Async()
- {
- WrappedEvent[] events_a_0 = new[] { new WrappedEvent(1) };
- WrappedEvent[] events_b_0 = new[] { new WrappedEvent(1) };
- WrappedEvent[] events_a_1 = new[] { new WrappedEvent(2) };
- WrappedEvent[] events_b_1 = new[] { new WrappedEvent(2) };
-
- await EventRepository.AppendEventsAsync(AggregateType, "a", events_a_0);
- await EventRepository.AppendEventsAsync(AggregateType, "b", events_b_0);
- await EventRepository.AppendEventsAsync(AggregateType, "a", events_a_1);
- await EventRepository.AppendEventsAsync(AggregateType, "b", events_b_1);
-
- IReadOnlyList<WrappedEvent> actualEventsA = await EventRepository.GetEventsAsync(AggregateType, "a");
- Assert.Equal(events_a_0.Concat(events_a_1), actualEventsA);
-
- IReadOnlyList<WrappedEvent> actualEventsB = await EventRepository.GetEventsAsync(AggregateType, "b");
- Assert.Equal(events_b_0.Concat(events_b_1), actualEventsB);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { "a", "b" }, actualAggregateIds);
- }
-
- [Fact]
- public async Task AppendEventsAsync_InterleavingConflict_Async()
- {
- WrappedEvent[] events_a_0 = new[] { new WrappedEvent(1) };
- WrappedEvent[] events_b_0 = new[] { new WrappedEvent(1) };
- WrappedEvent[] events_a_1 = new[] { new WrappedEvent(2) };
-
- async Task ActionAsync()
- {
- await EventRepository.AppendEventsAsync(AggregateType, "a", events_a_0).ConfigureAwait(false);
- await EventRepository.AppendEventsAsync(AggregateType, "b", events_b_0).ConfigureAwait(false);
- await EventRepository.AppendEventsAsync(AggregateType, "a", events_a_1).ConfigureAwait(false);
- await EventRepository.AppendEventsAsync(AggregateType, "a", events_a_1).ConfigureAwait(false);
- }
-
- await Assert.ThrowsAsync<OptimisticConcurrencyException>(ActionAsync);
-
- IReadOnlyList<WrappedEvent> actualEventsA = await EventRepository.GetEventsAsync(AggregateType, "a");
- Assert.Equal(events_a_0.Concat(events_a_1), actualEventsA);
-
- IReadOnlyList<WrappedEvent> actualEventsB = await EventRepository.GetEventsAsync(AggregateType, "b");
- Assert.Equal(events_b_0, actualEventsB);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { "a", "b" }, actualAggregateIds);
- }
-
- [Fact]
- public async Task AppendEvents_AppendIsAtomic_Async()
- {
- WrappedEvent[] events1 = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a") };
- WrappedEvent[] events2 = new[] { new TestWrappedEvent(2, "b"), new TestWrappedEvent(3, "b") };
-
- async Task ActionAsync()
- {
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events1).ConfigureAwait(false);
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events2).ConfigureAwait(false);
- }
-
- await Assert.ThrowsAsync<OptimisticConcurrencyException>(ActionAsync);
-
- IReadOnlyList<WrappedEvent> wrappedEvents = await EventRepository.GetEventsAsync(AggregateType, Id1);
- Assert.Equal(events1, wrappedEvents);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { Id1 }, actualAggregateIds);
- }
-
- [Fact]
- public async Task AppendEvents_CriticalSectionConflicts_Async()
- {
- WrappedEvent[] events1 = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a") };
- WrappedEvent[] events2 = new[] { new TestWrappedEvent(2, "b"), new TestWrappedEvent(3, "b") };
-
- async Task Append1Async()
- {
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events1);
- }
-
- async Task Append2Async()
- {
- bool entered = await TestEventRepository.AppendedSemaphore.WaitAsync(SemaphoreWaitTimeout).ConfigureAwait(false);
- Assert.True(entered, "Lock failed to be entered.");
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events2);
- }
-
- async Task AppendInParallelAsync()
- {
- Task task1 = Task.Run(Append1Async);
- Task task2 = Task.Run(Append2Async);
- await Task.WhenAll(task1, task2);
- }
-
- async Task WaitForConflictAsync()
- {
- bool entered = await TestEventRepository.ConflictedSemaphore.WaitAsync(SemaphoreWaitTimeout).ConfigureAwait(false);
- Assert.True(entered, "Lock failed to be entered.");
- }
-
- TestEventRepository.AppendedCallbackAsync = WaitForConflictAsync;
-
- await Assert.ThrowsAsync<OptimisticConcurrencyException>(AppendInParallelAsync);
- IReadOnlyList<WrappedEvent> enumerable = await EventRepository.GetEventsAsync(AggregateType, Id1);
- Assert.Equal(events1, enumerable);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { Id1 }, actualAggregateIds);
- }
-
- [Fact]
- public async Task AppendEvents_CriticalAppendConflicts_Async()
- {
- WrappedEvent[] events1 = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a") };
- WrappedEvent[] events2 = new[] { new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b") };
-
- async Task Append1Async()
- {
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events1).ConfigureAwait(false);
- }
-
- async Task Append2Async()
- {
- bool entered = await TestEventRepository.AppendedSemaphore.WaitAsync(SemaphoreWaitTimeout).ConfigureAwait(false);
- Assert.True(entered, "Lock failed to be entered.");
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events2).ConfigureAwait(false);
- }
-
- async Task AppendInParallelAsync()
- {
- await Task.WhenAll(Append1Async(), Append2Async()).ConfigureAwait(false);
- }
-
- async Task WaitForNoConflictAsync()
- {
- bool entered = await TestEventRepository.ConflictedSemaphore.WaitAsync(SemaphoreWaitTimeout).ConfigureAwait(false);
- Assert.False(entered, "Lock failed to be entered.");
- }
-
- TestEventRepository.AppendedCallbackAsync = WaitForNoConflictAsync;
-
- // No conflict.
- await AppendInParallelAsync();
-
- IReadOnlyList<WrappedEvent> actualEvents = await EventRepository.GetEventsAsync(AggregateType, Id1);
- Assert.Equal(events1.Concat(events2), actualEvents);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { Id1 }, actualAggregateIds);
- }
-
- [Theory]
- [InlineData(nameof(TestInMemoryEventRepository.ValidatedCallbackAsync))]
- [InlineData(nameof(TestInMemoryEventRepository.AppendedCallbackAsync))]
- public async Task ListEventsAsync_ConflictWithAppending_Async(string listOnCallback)
- {
- TestWrappedEvent[] events1 = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a") };
- TestWrappedEvent[] events2 = new[] { new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b") };
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events1);
-
- IReadOnlyList<WrappedEvent> result = Array.Empty<WrappedEvent>().ToList().AsReadOnly();
-
- Task ListCallbackAsync()
- {
- result = EventRepository.GetEventsAsync(AggregateType, Id1).Result;
- return Task.CompletedTask;
- }
-
- switch (listOnCallback)
- {
- case nameof(TestInMemoryEventRepository.ValidatedCallbackAsync):
- TestEventRepository.ValidatedCallbackAsync = ListCallbackAsync;
- break;
-
- case nameof(TestInMemoryEventRepository.AppendedCallbackAsync):
- TestEventRepository.AppendedCallbackAsync = ListCallbackAsync;
- break;
-
- default:
- throw new ApplicationException($"unexpected value listOnCallback: '{listOnCallback}'");
- }
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events2);
-
- IEnumerable<TestWrappedEvent> expected = events1.AsEnumerable();
-
- switch (listOnCallback)
- {
- case nameof(TestInMemoryEventRepository.AppendedCallbackAsync):
- expected = expected.Concat(events2);
- break;
- }
-
- Assert.Equal(expected, result);
- }
-
- [Theory]
- [InlineData(0, 1)]
- [InlineData(0, 2)]
- [InlineData(0, 3)]
- [InlineData(0, 4)]
- [InlineData(0, 5)]
- [InlineData(1, 1)]
- [InlineData(1, 2)]
- [InlineData(1, 3)]
- [InlineData(1, 4)]
- [InlineData(2, 1)]
- [InlineData(2, 2)]
- [InlineData(2, 3)]
- [InlineData(3, 1)]
- [InlineData(3, 2)]
- [InlineData(4, 1)]
- public async Task ListEventsAsync_OptionalArguments_Async(long afterSequenceId, int limit)
- {
- WrappedEvent[] events = new[]
- {
- new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a"),
- new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b")
- };
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events);
-
- var result = await EventRepository.GetEventsAsync(
- AggregateType, Id1, afterSequenceId, limit);
-
- Assert.True(result.Count <= limit);
- Assert.True(result.All(a => afterSequenceId < a.SequenceId));
- }
-
- [Fact]
- public async Task ListAggregateIdsAsync_Async()
- {
- WrappedEvent[] events = new[]
- {
- new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a"),
- new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b")
- };
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events);
- await EventRepository.AppendEventsAsync(AggregateType, "MY_ID_2", events);
-
- IReadOnlyList<string> actualAggregateIds = await EventRepository.GetAggregateIdsAsync(AggregateType);
- Assert.Equal(new string[] { Id1, "MY_ID_2" }, actualAggregateIds);
- }
-
- [Theory]
- [InlineData("0", 0)]
- [InlineData("0", 1)]
- [InlineData("0", 2)]
- [InlineData("0", 3)]
- [InlineData(Id1, 0)]
- [InlineData(Id1, 1)]
- [InlineData(Id1, 2)]
- [InlineData("MY_ID_2", 0)]
- [InlineData("MY_ID_2", 1)]
- [InlineData("3", 0)]
- [InlineData("3", 1)]
- public async Task ListAggregateIdsAsync_OptionalArguments_Async(string afterAggregateId, int limit)
- {
- WrappedEvent[] events = new[]
- {
- new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a"),
- new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b")
- };
-
- await EventRepository.AppendEventsAsync(AggregateType, Id1, events);
- await EventRepository.AppendEventsAsync(AggregateType, Id2, events);
-
- IReadOnlyList<string> result = await EventRepository.GetAggregateIdsAsync(AggregateType, afterAggregateId, limit);
-
- Assert.True(result.Count <= limit);
- Assert.True(result.All(a => afterAggregateId.CompareTo(a) <= 0));
- }
-
- public void Dispose()
- {
- TestEventRepository.Dispose();
- }
-}

DELETE
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestInMemoryEventRepository.cs
-using System.Threading;
-using System.Threading.Tasks;
-using WalletWasabi.EventSourcing;
-using Xunit.Abstractions;
-
-namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain;
-
-public class TestInMemoryEventRepository : InMemoryEventRepository, IDisposable
-{
- public TestInMemoryEventRepository(ITestOutputHelper testOutput)
- {
- TestOutput = testOutput;
- }
-
- protected ITestOutputHelper TestOutput { get; }
-
- public SemaphoreSlim ValidatedSemaphore { get; } = new(0);
- public SemaphoreSlim ConflictedSemaphore { get; } = new(0);
- public SemaphoreSlim AppendedSemaphore { get; } = new(0);
-
- public Func<Task>? ValidatedCallbackAsync { get; set; }
- public Func<Task>? ConflictedCallbackAsync { get; set; }
- public Func<Task>? AppendedCallbackAsync { get; set; }
-
- /// <summary>Helper method for verifying invariants in tests.</summary>
- protected override async Task ValidatedAsync()
- {
- await base.ValidatedAsync().ConfigureAwait(false);
-
- TestOutput.WriteLine(nameof(ValidatedAsync));
- ValidatedSemaphore.Release();
-
- if (ValidatedCallbackAsync is not null)
- {
- await ValidatedCallbackAsync().ConfigureAwait(false);
- }
- }
-
- /// <summary>Helper method for verifying invariants in tests.</summary>
- protected override async Task ConflictedAsync()
- {
- await base.ConflictedAsync().ConfigureAwait(false);
-
- TestOutput.WriteLine(nameof(ConflictedAsync));
- ConflictedSemaphore.Release();
-
- if (ConflictedCallbackAsync is not null)
- {
- await ConflictedCallbackAsync().ConfigureAwait(false);
- }
- }
-
- /// <summary>Helper method for verifying invariants in tests.</summary>
- protected override async Task AppendedAsync()
- {
- await base.AppendedAsync().ConfigureAwait(false);
-
- TestOutput.WriteLine(nameof(AppendedAsync));
- AppendedSemaphore.Release();
-
- if (AppendedCallbackAsync is not null)
- {
- await AppendedCallbackAsync().ConfigureAwait(false);
- }
- }
-
- public void Dispose()
- {
- ValidatedSemaphore.Dispose();
- ConflictedSemaphore.Dispose();
- AppendedSemaphore.Dispose();
- }
-}

DELETE
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestWrappedEvent.cs
-using WalletWasabi.EventSourcing;
-
-namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain;
-
-public record TestWrappedEvent(long SequenceId, string Value) : WrappedEvent(SequenceId);

DELETE
WalletWasabi/EventSourcing/AggregateEvents.cs
-using System.Collections.Immutable;
-
-namespace WalletWasabi.EventSourcing;
-
-public record AggregateEvents(long TailSequenceId, ImmutableList<WrappedEvent> Events)
-{
- /// <summary>
- /// SequenceId of the last event of this aggregate
- /// </summary>
- public long TailSequenceId { get; init; } = TailSequenceId;
-
- /// <summary>
- /// Ordered list of events
- /// </summary>
- public ImmutableList<WrappedEvent> Events { get; init; } = Events;
-}

DELETE
WalletWasabi/EventSourcing/AggregateTypeIds.cs
-using System;
-using System.Collections.Generic;
-using System.Collections.Immutable;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace WalletWasabi.EventSourcing;
-
-public record AggregateTypeIds(long TailIndex, ImmutableSortedSet<string> Ids)
-{
- /// <summary>
- /// Index of the last aggregateId in this aggregateType
- /// </summary>
- public long TailIndex { get; init; } = TailIndex;
-
- /// <summary>
- /// List of aggregate Ids in this aggregateType
- /// </summary>
- public ImmutableSortedSet<string> Ids { get; init; } = Ids;
-}

DELETE
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
-using System.Collections.Concurrent;
-using System.Collections.Generic;
-using System.Collections.Immutable;
-using System.Collections.ObjectModel;
-using System.Linq;
-using System.Threading.Tasks;
-using WalletWasabi.Exceptions;
-using WalletWasabi.Interfaces.EventSourcing;
-
-namespace WalletWasabi.EventSourcing;
-
-/// <summary>
-/// Thread safe without locks in memory event repository implementation
-/// </summary>
-public class InMemoryEventRepository : IEventRepository
-{
- private static readonly IReadOnlyList<WrappedEvent> EmptyResult = ImmutableList<WrappedEvent>.Empty;
-
- private static readonly IReadOnlyList<string> EmptyIds = ImmutableList<string>.Empty;
-
- private static readonly IComparer<WrappedEvent> WrappedEventSequenceIdComparer
- = Comparer<WrappedEvent>.Create((a, b) => a.SequenceId.CompareTo(b.SequenceId));
-
- private ConcurrentDictionary
- // aggregateType
- <string,
- ConcurrentDictionary
- // aggregateId
- <string, AggregateEvents>> AggregatesEventsBatches
- { get; } = new();
-
- private ConcurrentDictionary
- // aggregateType
- <string,
- AggregateTypeIds> AggregatesIds
- { get; } = new();
-
- /// <inheritdoc/>
- public async Task AppendEventsAsync(string aggregateType, string aggregateId, IEnumerable<WrappedEvent> wrappedEvents)
- {
- ReadOnlyCollection<WrappedEvent> wrappedEventsList = wrappedEvents.ToList().AsReadOnly();
-
- if (wrappedEventsList.Count == 0)
- {
- return;
- }
-
- long firstSequenceId = wrappedEventsList[0].SequenceId;
- long lastSequenceId = wrappedEventsList[^1].SequenceId;
-
- if (firstSequenceId <= 0)
- {
- throw new ArgumentException("First event sequenceId is not a positive number.", nameof(wrappedEvents));
- }
-
- if (lastSequenceId <= 0)
- {
- throw new ArgumentException("Last event sequenceId is not a positive integer.", nameof(wrappedEvents));
- }
-
- if (lastSequenceId - firstSequenceId + 1 != wrappedEventsList.Count)
- {
- throw new ArgumentException("Event sequence IDs are inconsistent.", nameof(wrappedEvents));
- }
-
- var aggregateEventsBatches = AggregatesEventsBatches.GetOrAdd(aggregateType, _ => new());
- var (tailSequenceId, events) = aggregateEventsBatches.GetOrAdd(
- aggregateId,
- _ => new AggregateEvents(0, ImmutableList<WrappedEvent>.Empty));
-
- if (tailSequenceId + 1 < firstSequenceId)
- {
- throw new ArgumentException($"Invalid firstSequenceId (gap in sequence IDs) expected: '{tailSequenceId + 1}' given: '{firstSequenceId}'.", nameof(wrappedEvents));
- }
-
- // No action.
- await ValidatedAsync().ConfigureAwait(false);
-
- ImmutableList<WrappedEvent> newEvents = events.AddRange(wrappedEventsList);
-
- // Atomically detect conflict and replace lastSequenceId and lock to ensure strong order in eventsBatches.
- if (!aggregateEventsBatches.TryUpdate(
- key: aggregateId,
- newValue: new AggregateEvents(lastSequenceId, newEvents),
- comparisonValue: new AggregateEvents(firstSequenceId - 1, events)))
- {
- await ConflictedAsync().ConfigureAwait(false);
-
- throw new OptimisticConcurrencyException($"Conflict while committing events. Retry command. aggregate: '{aggregateType}' ID: '{aggregateId}'");
- }
-
- await AppendedAsync().ConfigureAwait(false);
-
- // If it is a first event for given aggregate.
- if (tailSequenceId == 0)
- {
- // Add index of aggregate id into the dictionary.
- IndexNewAggregateId(aggregateType, aggregateId);
- }
- }
-
- /// <remarks>
- /// Working with <see cref="ImmutableList{T}.BinarySearch(int, int, T, IComparer{T}?)"/> is explained here
- /// https://docs.microsoft.com/en-us/dotnet/api/system.collections.immutable.immutablelist-1.binarysearch.
- /// </remarks>
- /// <inheritdoc/>
- public Task<IReadOnlyList<WrappedEvent>> GetEventsAsync(string aggregateType, string aggregateId, long afterSequenceId = 0, int? maxCount = null)
- {
- if (AggregatesEventsBatches.TryGetValue(aggregateType, out ConcurrentDictionary<string, AggregateEvents>? aggregateEventsBatches) &&
- aggregateEventsBatches.TryGetValue(aggregateId, out AggregateEvents? value))
- {
- ImmutableList<WrappedEvent> result = value.Events;
-
- if (afterSequenceId > 0)
- {
- int foundIndex = result.BinarySearch(new WrappedEvent(afterSequenceId), WrappedEventSequenceIdComparer);
-
- if (foundIndex < 0)
- {
- foundIndex = ~foundIndex;
- }
- else
- {
- foundIndex++;
- }
-
- result = result.GetRange(foundIndex, result.Count - foundIndex);
- }
-
- if (maxCount < result.Count)
- {
- result = result.GetRange(0, maxCount.Value);
- }
-
- return Task.FromResult((IReadOnlyList<WrappedEvent>)result);
- }
-
- return Task.FromResult(EmptyResult);
- }
-
- /// <inheritdoc/>
- public Task<IReadOnlyList<string>> GetAggregateIdsAsync(string aggregateType, string? afterAggregateId = null, int? maxCount = null)
- {
- if (AggregatesIds.TryGetValue(aggregateType, out var aggregateIds))
- {
- ImmutableSortedSet<string> ids = aggregateIds.Ids;
- int foundIndex = 0;
-
- if (afterAggregateId != null)
- {
- foundIndex = ids.IndexOf(afterAggregateId);
- if (foundIndex < 0)
- {
- foundIndex = ~foundIndex;
- }
- else
- {
- foundIndex++;
- }
- }
-
- List<string> result = new();
- int afterLastIndex = maxCount.HasValue
- ? Math.Min(foundIndex + maxCount.Value, ids.Count)
- : ids.Count;
-
- for (int i = foundIndex; i < afterLastIndex; i++)
- {
- result.Add(ids[i]);
- }
-
- return Task.FromResult((IReadOnlyList<string>)result.AsReadOnly());
- }
-
- return Task.FromResult(EmptyIds);
- }
-
- private void IndexNewAggregateId(string aggregateType, string aggregateId)
- {
- ImmutableSortedSet<string> aggregateIds;
- ImmutableSortedSet<string> newAggregateIds;
-
- long tailIndex = 0L;
- int liveLockLimit = 10000;
-
- do
- {
- if (liveLockLimit <= 0)
- {
- throw new ApplicationException("Live lock detected.");
- }
-
- liveLockLimit--;
- (tailIndex, aggregateIds) = AggregatesIds.GetOrAdd(aggregateType, _ => new(0, ImmutableSortedSet<string>.Empty));
- newAggregateIds = aggregateIds.Add(aggregateId);
- }
- while (!AggregatesIds.TryUpdate(
- key: aggregateType,
- newValue: new AggregateTypeIds(tailIndex + 1, newAggregateIds),
- comparisonValue: new AggregateTypeIds(tailIndex, aggregateIds)));
- }
-
- /// <summary>Helper method for verifying invariants in tests.</summary>
- /// <remarks>Do not add any code to this method.</remarks>
- protected virtual Task ValidatedAsync()
- {
- return Task.CompletedTask;
- }
-
- /// <summary>Helper method for verifying invariants in tests.</summary>
- /// <remarks>Do not add any code to this method.</remarks>
- protected virtual Task ConflictedAsync()
- {
- return Task.CompletedTask;
- }
-
- /// <summary>Helper method for verifying invariants in tests.</summary>
- /// <remarks>Do not add any code to this method.</remarks>
- protected virtual Task AppendedAsync()
- {
- return Task.CompletedTask;
- }
-}

DELETE
WalletWasabi/EventSourcing/WrappedEvent.cs
-namespace WalletWasabi.EventSourcing;
-
-/// <param name="SequenceId">Only positive integers are allowed.</param>
-public record WrappedEvent(long SequenceId);

DELETE
WalletWasabi/Interfaces/EventSourcing/IEventRepository.cs
-using System.Collections.Generic;
-using System.Threading.Tasks;
-using WalletWasabi.EventSourcing;
-using WalletWasabi.Exceptions;
-
-namespace WalletWasabi.Interfaces.EventSourcing;
-
-public interface IEventRepository
-{
- /// <summary>
- /// Atomically persistently appends ordered list of events for given aggregate.
- /// In case of duplicate of <seealso cref="WrappedEvent.SequenceId"/>
- /// <seealso cref="OptimisticConcurrencyException"/> is thrown indicating that
- /// command should be retried with freshly loaded events from <see cref="GetEventsAsync"/>.
- /// </summary>
- /// <param name="aggregateType">Type of aggregate</param>
- /// <param name="aggregateId">ID of aggregate</param>
- /// <param name="wrappedEvents">Ordered list of events to be persisted</param>
- /// <exception cref="OptimisticConcurrencyException">If there is concurrency conflict; retry command</exception>
- /// <exception cref="TransientException">Transient infrastructure failure</exception>
- /// <exception cref="ArgumentException">Invalid input</exception>
- public Task AppendEventsAsync(string aggregateType, string aggregateId, IEnumerable<WrappedEvent> wrappedEvents);
-
- /// <summary>
- /// List strongly ordered events of given aggregate. This is the primary source of truth.
- /// Events are used to reconstruct aggregate state before processing command.
- /// </summary>
- /// <param name="aggregateType">Type of aggregate</param>
- /// <param name="aggregateId">ID of aggregate</param>
- /// <param name="afterSequenceId">Starts with event after given <seealso cref="WrappedEvent.SequenceId"/> and lists all following events</param>
- /// <param name="maxCount">Limits the number of returned events</param>
- /// <returns>Ordered list of events</returns>
- /// <exception cref="TransientException">Transient infrastructure failure</exception>
- public Task<IReadOnlyList<WrappedEvent>> GetEventsAsync(string aggregateType, string aggregateId, long afterSequenceId = 0, int? maxCount = null);
-
- /// <summary>
- /// Supplementary method for enumerating all ids for <paramref name="aggregateType"/>
- /// in this event repository. Order of ids is not defined can be any artificial.
- /// </summary>
- /// <param name="aggregateType">Type of aggregate</param>
- /// <param name="afterAggregateId">Starts with id right after given ID and lists all following IDs in any artificial order</param>
- /// <param name="maxCount">Limits the number of returned events</param>
- /// <returns>Unordered list of IDs</returns>
- /// <exception cref="TransientException">Transient infrastructure failure</exception>
- public Task<IReadOnlyList<string>> GetAggregateIdsAsync(string aggregateType, string? afterAggregateId = null, int? maxCount = null);
-}

Remove EventSourcing code