Merge pull request #54 from atc-net/feature/delete-stream

Implement DeleteStream in Management Client

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+###  Added
+
+-   Implement `IEventStoreManagementClient.DeleteStreamAsync` using the newly released `DeleteAllItemsByPartitionKeyStreamAsync` method in the Cosmos SDK.
+
 ## [1.13.3] - 2024-04-21
 
 ### Added
@@ -187,4 +191,4 @@ services.AddEventStore(builder =>
 
 [1.3.3]: https://github.com/atc-net/atc-cosmos-eventstore/compare/v1.2.9...v1.3.3
 
-[1.2.9]: https://github.com/atc-net/atc-cosmos-eventstore/compare/v1.1.3...v1.2.9
+[1.2.9]: https://github.com/atc-net/atc-cosmos-eventstore/compare/v1.1.3...v1.2.9

src/Atc.Cosmos.EventStore/Cosmos/CosmosDeleter.cs

@@ -0,0 +1,25 @@
+using Atc.Cosmos.EventStore.Streams;
+using Microsoft.Azure.Cosmos;
+
+namespace Atc.Cosmos.EventStore.Cosmos;
+
+internal class CosmosDeleter : IStreamDeleter
+{
+    private readonly IEventStoreContainerProvider containerProvider;
+
+    public CosmosDeleter(IEventStoreContainerProvider containerProvider)
+    {
+        this.containerProvider = containerProvider;
+    }
+
+    public async Task DeleteAsync(
+        StreamId streamId,
+        CancellationToken cancellationToken)
+    {
+        var pk = new PartitionKey(streamId.Value);
+        var container = containerProvider.GetStreamContainer();
+        var response = await container.DeleteAllItemsByPartitionKeyStreamAsync(pk, null, cancellationToken);
+
+        response.EnsureSuccessStatusCode();
+    }
+}

src/Atc.Cosmos.EventStore/DependencyInjection/EventStoreOptionsBuilder.cs

@@ -33,6 +33,7 @@ public EventStoreOptionsBuilder UseCosmosDb(
         Services.TryAddSingleton<IStreamMetadataReader, CosmosMetadataReader>();
         Services.TryAddSingleton<IStreamIterator, CosmosStreamIterator>();
         Services.TryAddSingleton<IStreamBatchWriter, CosmosBatchWriter>();
+        Services.TryAddSingleton<IStreamDeleter, CosmosDeleter>();
         Services.TryAddSingleton<IStreamSubscriptionFactory, CosmosSubscriptionFactory>();
 
         Services.TryAddSingleton<IStreamSubscriptionRemover, CosmosSubscriptionRemover>();
@@ -74,6 +75,7 @@ internal EventStoreOptionsBuilder UseInMemoryDb()
         Services.TryAddSingleton<IStreamMetadataReader>(s => s.GetRequiredService<InMemoryStore>());
         Services.TryAddSingleton<IStreamIterator>(s => s.GetRequiredService<InMemoryStore>());
         Services.TryAddSingleton<IStreamBatchWriter>(s => s.GetRequiredService<InMemoryStore>());
+        Services.TryAddSingleton<IStreamDeleter>(s => s.GetRequiredService<InMemoryStore>());
         Services.TryAddSingleton<IStreamSubscriptionFactory>(s => s.GetRequiredService<InMemoryStore>());
         Services.TryAddSingleton<IStreamSubscriptionRemover>(s => s.GetRequiredService<InMemoryStore>());
         Services.TryAddSingleton<IStreamIndexReader>(s => s.GetRequiredService<InMemoryStore>());

src/Atc.Cosmos.EventStore/EventStoreManagementClient.cs

@@ -1,11 +1,20 @@
+using Atc.Cosmos.EventStore.Streams;
+
 namespace Atc.Cosmos.EventStore;
 
 internal class EventStoreManagementClient : IEventStoreManagementClient
 {
+    private readonly IStreamDeleter streamDeleter;
+
+    public EventStoreManagementClient(IStreamDeleter streamDeleter)
+    {
+        this.streamDeleter = streamDeleter;
+    }
+
     public Task DeleteStreamAsync(
         StreamId streamId,
         CancellationToken cancellationToken = default)
-        => throw new NotImplementedException();
+        => streamDeleter.DeleteAsync(streamId, cancellationToken);
 
     public Task PurgeStreamAsync(
         StreamId streamId,

src/Atc.Cosmos.EventStore/IEventStoreManagementClient.cs

@@ -39,7 +39,7 @@ Task PurgeStreamAsync(
         CancellationToken cancellationToken = default);
 
     /// <summary>
-    /// Deletes an entire stream and it's index.
+    /// Deletes an entire stream.
     /// </summary>
     /// <remarks>Attempting to write to a deleted stream will create a new empty stream.</remarks>
     /// <param name="streamId">Id of the event stream to delete.</param>

src/Atc.Cosmos.EventStore/InMemory/InMemoryStore.cs

@@ -9,6 +9,7 @@ internal class InMemoryStore :
     IStreamMetadataReader,
     IStreamIterator,
     IStreamBatchWriter,
+    IStreamDeleter,
     IStreamSubscriptionFactory,
     IStreamSubscriptionRemover,
     IStreamIndexReader,
@@ -69,6 +70,9 @@ Task<IStreamMetadata> IStreamBatchWriter.WriteAsync(
         CancellationToken cancellationToken)
         => throw new NotImplementedException();
 
+    public Task DeleteAsync(StreamId streamId, CancellationToken cancellationToken)
+        => throw new NotImplementedException();
+
     public Task WriteAsync(
         string name,
         StreamId streamId,

src/Atc.Cosmos.EventStore/Streams/IStreamDeleter.cs

@@ -0,0 +1,8 @@
+namespace Atc.Cosmos.EventStore.Streams;
+
+internal interface IStreamDeleter
+{
+    Task DeleteAsync(
+        StreamId streamId,
+        CancellationToken cancellationToken);
+}

test/Atc.Cosmos.EventStore.Tests/Cosmos/CosmosDeleterTests.cs

@@ -0,0 +1,50 @@
+using Atc.Cosmos.EventStore.Cosmos;
+using Atc.Test;
+using Microsoft.Azure.Cosmos;
+using NSubstitute;
+using Xunit;
+
+namespace Atc.Cosmos.EventStore.Tests.Cosmos;
+
+public class CosmosDeleterTests
+{
+    private readonly ResponseMessage responseMessage;
+    private readonly Container container;
+    private readonly IEventStoreContainerProvider containerProvider;
+    private readonly CosmosDeleter sut;
+
+    public CosmosDeleterTests()
+    {
+        responseMessage = Substitute.For<ResponseMessage>();
+        responseMessage.IsSuccessStatusCode.Returns(true);
+
+        container = Substitute.For<Container>();
+        container
+            .DeleteAllItemsByPartitionKeyStreamAsync(default, default, default)
+            .ReturnsForAnyArgs(responseMessage);
+
+        containerProvider = Substitute.For<IEventStoreContainerProvider>();
+        containerProvider
+            .GetStreamContainer()
+            .Returns(container, returnThese: null);
+
+        sut = new CosmosDeleter(containerProvider);
+    }
+
+    [Theory, AutoNSubstituteData]
+    public async Task Should_Use_StreamId_As_PartitionKey(
+        StreamId streamId,
+        CancellationToken cancellationToken)
+    {
+        await sut.DeleteAsync(
+            streamId,
+            cancellationToken);
+
+        _ = container
+            .Received()
+            .DeleteAllItemsByPartitionKeyStreamAsync(
+                new PartitionKey(streamId.Value),
+                null,
+                cancellationToken);
+    }
+}

test/Atc.Cosmos.EventStore.Tests/EventStoreClientTests.cs

@@ -1,6 +1,5 @@
 using System.Collections.ObjectModel;
 using Atc.Cosmos.EventStore.Cosmos;
-using Atc.Cosmos.EventStore.Events;
 using Atc.Cosmos.EventStore.Streams;
 using Atc.Test;
 using AutoFixture.AutoNSubstitute;
@@ -104,7 +103,7 @@ await sut
     }
 
     [Theory, AutoNSubstituteData]
-    internal async Task Should_Throw_When_EventsList_Containes_NullObject(
+    internal async Task Should_Throw_When_EventsList_Contains_NullObject(
         EventStoreClient sut,
         StreamId streamId,
         Collection<object> events,

test/Atc.Cosmos.EventStore.Tests/EventStoreManagementClientTests.cs

@@ -0,0 +1,29 @@
+using Atc.Cosmos.EventStore.Streams;
+using Atc.Test;
+using AutoFixture.Xunit2;
+using FluentAssertions;
+using NSubstitute;
+using Xunit;
+
+namespace Atc.Cosmos.EventStore.Tests;
+
+public class EventStoreManagementClientTests
+{
+    [Theory, AutoNSubstituteData]
+    internal async Task Should_DeleteStream(
+        [Frozen] IStreamDeleter deleter,
+        EventStoreManagementClient sut,
+        StreamId streamId,
+        CancellationToken cancellationToken)
+    {
+        await sut.DeleteStreamAsync(
+            streamId,
+            cancellationToken: cancellationToken);
+
+        _ = deleter
+            .Received(1)
+            .DeleteAsync(
+                streamId,
+                cancellationToken);
+    }
+}