Feature/low level stream

Directory.Packages.props

@@ -1,58 +1,59 @@
 <Project>
-  <PropertyGroup>
-    <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
-    <CentralPackageTransitivePinningEnabled>true</CentralPackageTransitivePinningEnabled>
-  </PropertyGroup>
-  <ItemGroup>
-    <PackageVersion Include="AutoFixture.Xunit2" Version="4.18.1" />
-    <PackageVersion Include="Azure.Core" Version="1.53.0" />
-    <PackageVersion Include="Azure.Identity" Version="[1.15.0, 2.0]" />
-    <PackageVersion Include="coverlet.collector" Version="8.0.1">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageVersion>
-    <PackageVersion Include="JsonSchema.Net" Version="9.1.4" />
-    <PackageVersion Include="Microsoft.Azure.Cosmos" Version="3.58.0" />
-    <PackageVersion Include="Nerdbank.GitVersioning" Version="3.9.50">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageVersion>
-    <PackageVersion Include="Microsoft.CodeAnalysis" Version="5.0.0" />
-    <PackageVersion Include="Microsoft.CodeAnalysis.CSharp.SourceGenerators.Testing.XUnit" Version="1.1.2" />
-    <PackageVersion Include="Microsoft.CodeAnalysis.CSharp" Version="5.0.0" />
-    <PackageVersion Include="Microsoft.CodeAnalysis.Analyzers" Version="3.11.0" />
-    <PackageVersion Include="Microsoft.EntityFrameworkCore" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.5">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageVersion>
-    <PackageVersion Include="Microsoft.EntityFrameworkCore.InMemory" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.EntityFrameworkCore.SqlServer" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.Configuration" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.Configuration.Binder" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.Extensions.Options" Version="10.0.5" />
-    <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.4.0" />
-    <PackageVersion Include="MongoDB.Driver" Version="3.7.1" />
-    <PackageVersion Include="Moq" Version="4.20.72" />
-    <PackageVersion Include="Newtonsoft.Json" Version="13.0.4" />
-    <PackageVersion Include="Papst.EventStore" Version="[6.0.0,7.0)" />
-    <PackageVersion Include="Shouldly" Version="4.3.0" />
-    <PackageVersion Include="Swashbuckle.AspNetCore" Version="10.0.1" />
-    <PackageVersion Include="System.Formats.Asn1" Version="10.0.1" />
-    <PackageVersion Include="System.Linq.Async" Version="7.0.0" />
-    <PackageVersion Include="Testcontainers" Version="4.11.0" />
-    <PackageVersion Include="Testcontainers.CosmosDb" Version="4.11.0" />
-    <PackageVersion Include="Testcontainers.MongoDb" Version="4.11.0" />
-    <PackageVersion Include="xunit" Version="2.9.3" />
-    <PackageVersion Include="xunit.runner.visualstudio" Version="3.1.5">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageVersion>
-  </ItemGroup>
+    <PropertyGroup>
+        <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
+        <CentralPackageTransitivePinningEnabled>true</CentralPackageTransitivePinningEnabled>
+    </PropertyGroup>
+    <ItemGroup>
+        <PackageVersion Include="AutoFixture.Xunit2" Version="4.18.1" />
+        <PackageVersion Include="Azure.Core" Version="1.53.0" />
+        <PackageVersion Include="Azure.Identity" Version="[1.15.0, 2.0]" />
+        <PackageVersion Include="coverlet.collector" Version="8.0.1">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageVersion>
+        <PackageVersion Include="JsonSchema.Net" Version="9.1.4" />
+        <PackageVersion Include="Microsoft.Azure.Cosmos" Version="3.58.0" />
+        <PackageVersion Include="Nerdbank.GitVersioning" Version="3.9.50">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageVersion>
+        <PackageVersion Include="Microsoft.CodeAnalysis" Version="5.0.0" />
+        <PackageVersion Include="Microsoft.CodeAnalysis.CSharp.SourceGenerators.Testing.XUnit" Version="1.1.2" />
+        <PackageVersion Include="Microsoft.CodeAnalysis.CSharp" Version="5.0.0" />
+        <PackageVersion Include="Microsoft.CodeAnalysis.Analyzers" Version="3.11.0" />
+        <PackageVersion Include="Microsoft.EntityFrameworkCore" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.5">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageVersion>
+        <PackageVersion Include="Microsoft.EntityFrameworkCore.InMemory" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.EntityFrameworkCore.SqlServer" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.Configuration" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.Configuration.Binder" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.Extensions.Options" Version="10.0.5" />
+        <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.4.0" />
+        <PackageVersion Include="MongoDB.Driver" Version="3.8.1" />
+        <PackageVersion Include="Moq" Version="4.20.72" />
+        <PackageVersion Include="Newtonsoft.Json" Version="13.0.4" />
+        <PackageVersion Include="Papst.EventStore" Version="[6.0.0,7.0)" />
+        <PackageVersion Include="SharpCompress" Version="0.48.1" />
+        <PackageVersion Include="Shouldly" Version="4.3.0" />
+        <PackageVersion Include="Swashbuckle.AspNetCore" Version="10.0.1" />
+        <PackageVersion Include="System.Formats.Asn1" Version="10.0.1" />
+        <PackageVersion Include="System.Linq.Async" Version="7.0.0" />
+        <PackageVersion Include="Testcontainers" Version="4.11.0" />
+        <PackageVersion Include="Testcontainers.CosmosDb" Version="4.11.0" />
+        <PackageVersion Include="Testcontainers.MongoDb" Version="4.11.0" />
+        <PackageVersion Include="xunit" Version="2.9.3" />
+        <PackageVersion Include="xunit.runner.visualstudio" Version="3.1.5">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageVersion>
+    </ItemGroup>
 </Project>

README.md

@@ -59,6 +59,38 @@ Please refer to the documentation in the relevant implementation sources:
 * [Azure Cosmos](./src/Papst.EventStore.AzureCosmos/README.md)
 * [Entity Framework Core](./src/Papst.EventStore.EntityFrameworkCore/README.md)
 
+## Low-level event access
+
+The `ILowLevelEventStream` interface provides a low-level append API that accepts a raw `JObject` together with an explicit event type name:
+
+```csharp
+Task AppendAsync(
+    Guid id,
+    string eventType,
+    JObject evt,
+    EventStreamMetaData? metaData = null,
+    CancellationToken cancellationToken = default);
+```
+
+This API is currently only implemented by the Azure Cosmos provider.
+
+`ILowLevelEventStream` is not resolved directly from `IEventStore`. Instead, obtain an `IEventStream` first and then cast it to `ILowLevelEventStream` when the underlying provider supports it:
+
+```csharp
+IEventStream stream = await eventStore.GetAsync(streamId, cancellationToken);
+
+if (stream is ILowLevelEventStream lowLevelStream)
+{
+    await lowLevelStream.AppendAsync(
+        Guid.NewGuid(),
+        "MyExternalEvent",
+        JObject.Parse("""{ \"value\": 42 }"""),
+        cancellationToken: cancellationToken);
+}
+```
+
+This is intended for scenarios where the event payload or event type is not known at compile time. If you work with strongly typed events, prefer `IEventStream.AppendAsync<TEvent>()`.
+
 ## Event Catalog
 
 The **Event Catalog** provides a queryable registry of all events associated with a given entity type, including metadata (description, constraints) and a compile-time generated JSON Schema. This is useful for documentation, API discovery, and runtime introspection.

src/Papst.EventStore.AzureCosmos/CosmosEventStream.cs

@@ -18,7 +18,7 @@ internal sealed class CosmosEventStream(
   ICosmosIdStrategy idStrategy,
   TimeProvider timeProvider
 )
-  : IEventStream
+  : IEventStream, ILowLevelEventStream
 {
   private const int MaxBatchSize = 100;
 
@@ -81,6 +81,26 @@ TimeProvider timeProvider
     },
   };
 
+  public async Task AppendAsync(Guid id, string eventType, JObject evt, EventStreamMetaData? metaData = null, CancellationToken cancellationToken = default)
+  {
+    EventStreamDocumentEntity document = new()
+    {
+      Id = await idStrategy.GenerateIdAsync(StreamId, _stream.NextVersion, EventStreamDocumentType.Event),
+      DocumentId = id,
+      StreamId = StreamId,
+      Version = _stream.NextVersion,
+      Data = JObject.FromObject(evt),
+      DataType = eventType,
+      Name = eventType,
+      Time = timeProvider.GetLocalNow(),
+      DocumentType = EventStreamDocumentType.Event,
+      MetaData = metaData ?? new(),
+      TargetType = _stream.TargetType,
+    };
+
+    await AppendDocumentAsync(document, metaData, cancellationToken).ConfigureAwait(false);
+  }
+
   public async Task AppendAsync<TEvent>(
     Guid id,
     TEvent evt,
@@ -90,52 +110,7 @@ public async Task AppendAsync<TEvent>(
   {
     string eventName = eventTypeProvider.ResolveType(typeof(TEvent));
     EventStreamDocumentEntity document = await CreateEventEntity(id, evt, metaData, eventName).ConfigureAwait(false);
-    bool indexUpdateSuccessful = false;
-    int retryCount = 0;
-    do
-    {
-      try
-      {
-        List<PatchOperation> patches =
-        [
-          PatchOperation.Set('/' + nameof(EventStreamIndexEntity.NextVersion), _stream.NextVersion + 1),
-          PatchOperation.Set('/' + nameof(EventStreamIndexEntity.Version), _stream.NextVersion),
-          PatchOperation.Set('/' + nameof(EventStreamIndexEntity.Updated), timeProvider.GetLocalNow()),
-        ];
-        if (metaData is not null && options.UpdateTenantIdOnAppend && metaData.TenantId is not null)
-        {
-          patches.Add(PatchOperation.Set(
-            '/' + nameof(EventStreamIndexEntity.MetaData) + '/' + nameof(EventStreamMetaData.TenantId),
-            metaData.TenantId)
-          );
-        }
-
-        ItemResponse<EventStreamIndexEntity> indexPatch = await dbProvider.Container
-          .PatchItemAsync<EventStreamIndexEntity>(
-            _stream.Id,
-            new(StreamId.ToString()),
-            patches,
-            new PatchItemRequestOptions() { IfMatchEtag = _stream.ETag },
-            cancellationToken).ConfigureAwait(false);
-
-        _stream = indexPatch.Resource;
-        indexUpdateSuccessful = true;
-      }
-      catch (CosmosException e)
-      {
-        logger.IndexPatchConcurrency(e, _stream.StreamId);
-        retryCount++;
-        await Task.Delay(TimeSpan.FromMilliseconds(9), cancellationToken).ConfigureAwait(false);
-        await RefreshIndexAsync(cancellationToken).ConfigureAwait(false);
-      }
-    } while (!indexUpdateSuccessful && retryCount < options.ConcurrencyRetryCount);
-
-    logger.AppendingEvent(document.DataType, document.StreamId, document.Version);
-    _ = await dbProvider.Container.CreateItemAsync(
-        document,
-        new(StreamId.ToString()),
-        cancellationToken: cancellationToken)
-      .ConfigureAwait(false);
+    await AppendDocumentAsync(document, metaData, cancellationToken).ConfigureAwait(false);
   }
 
   public async Task AppendSnapshotAsync<TEntity>(
@@ -149,31 +124,37 @@ public async Task AppendSnapshotAsync<TEntity>(
     string eventName = typeof(TEntity).Name;
     EventStreamDocumentEntity document =
       await CreateEventEntity(id, entity, metaData, eventName, EventStreamDocumentType.Snapshot).ConfigureAwait(false);
+    await AppendDocumentAsync(
+        document,
+        metaData,
+        cancellationToken,
+        PatchOperation.Set('/' + nameof(EventStreamIndexEntity.LatestSnapshotVersion), _stream.NextVersion))
+      .ConfigureAwait(false);
+  }
+
+  private async Task RefreshIndexAsync(CancellationToken cancellationToken) => _stream = await dbProvider.Container
+    .ReadItemAsync<EventStreamIndexEntity>(_stream.Id, new(StreamId.ToString()), cancellationToken: cancellationToken)
+    .ConfigureAwait(false);
+
+  private async Task AppendDocumentAsync(
+    EventStreamDocumentEntity document,
+    EventStreamMetaData? metaData,
+    CancellationToken cancellationToken,
+    params PatchOperation[] additionalPatches
+  )
+  {
     bool indexUpdateSuccessful = false;
     int retryCount = 0;
     do
     {
       try
       {
-        List<PatchOperation> patches =
-        [
-          PatchOperation.Set('/' + nameof(EventStreamIndexEntity.NextVersion), _stream.NextVersion + 1),
-          PatchOperation.Set('/' + nameof(EventStreamIndexEntity.Version), _stream.NextVersion),
-          PatchOperation.Set('/' + nameof(EventStreamIndexEntity.Updated), DateTimeOffset.Now),
-          PatchOperation.Set('/' + nameof(EventStreamIndexEntity.LatestSnapshotVersion), _stream.NextVersion),
-        ];
-        if (metaData is not null && options.UpdateTenantIdOnAppend && metaData.TenantId is not null)
-        {
-          patches.Add(PatchOperation.Set(
-            '/' + nameof(EventStreamIndexEntity.MetaData) + '/' + nameof(EventStreamMetaData.TenantId),
-            metaData.TenantId)
-          );
-        }
+        List<PatchOperation> patches = CreateAppendPatches(metaData, additionalPatches);
 
         ItemResponse<EventStreamIndexEntity> indexPatch = await dbProvider.Container
           .PatchItemAsync<EventStreamIndexEntity>(
             _stream.Id,
-            new PartitionKey(StreamId.ToString()),
+            new(StreamId.ToString()),
             patches,
             new PatchItemRequestOptions() { IfMatchEtag = _stream.ETag },
             cancellationToken).ConfigureAwait(false);
@@ -193,14 +174,34 @@ public async Task AppendSnapshotAsync<TEntity>(
     logger.AppendingEvent(document.DataType, document.StreamId, document.Version);
     _ = await dbProvider.Container.CreateItemAsync(
         document,
-        new PartitionKey(StreamId.ToString()),
+        new(StreamId.ToString()),
         cancellationToken: cancellationToken)
       .ConfigureAwait(false);
   }
 
-  private async Task RefreshIndexAsync(CancellationToken cancellationToken) => _stream = await dbProvider.Container
-    .ReadItemAsync<EventStreamIndexEntity>(_stream.Id, new(StreamId.ToString()), cancellationToken: cancellationToken)
-    .ConfigureAwait(false);
+  private List<PatchOperation> CreateAppendPatches(
+    EventStreamMetaData? metaData,
+    params PatchOperation[] additionalPatches
+  )
+  {
+    List<PatchOperation> patches =
+    [
+      PatchOperation.Set('/' + nameof(EventStreamIndexEntity.NextVersion), _stream.NextVersion + 1),
+      PatchOperation.Set('/' + nameof(EventStreamIndexEntity.Version), _stream.NextVersion),
+      PatchOperation.Set('/' + nameof(EventStreamIndexEntity.Updated), timeProvider.GetLocalNow()),
+    ];
+    if (metaData is not null && options.UpdateTenantIdOnAppend && metaData.TenantId is not null)
+    {
+      patches.Add(PatchOperation.Set(
+        '/' + nameof(EventStreamIndexEntity.MetaData) + '/' + nameof(EventStreamMetaData.TenantId),
+        metaData.TenantId)
+      );
+    }
+
+    patches.AddRange(additionalPatches);
+
+    return patches;
+  }
 
   private async ValueTask<EventStreamDocumentEntity> CreateEventEntity<TEvent>(
     Guid id,
@@ -532,7 +533,7 @@ private async IAsyncEnumerable<TransactionalBatch> CreateBatches(IReadOnlyList<E
     {
       var version = currentVersion + (ulong)i + 1;
 
-      currentBatch.CreateItem(new EventStreamDocumentEntity
+      currentBatch = currentBatch.CreateItem(new EventStreamDocumentEntity
       {
         Id = await idStrategy.GenerateIdAsync(StreamId, version, EventStreamDocumentType.Event),
         DocumentId = events[i].DocumentId,
@@ -561,4 +562,5 @@ private async IAsyncEnumerable<TransactionalBatch> CreateBatches(IReadOnlyList<E
       yield return currentBatch;
     }
   }
+
 }

src/Papst.EventStore/ILowLevelEventStream.cs

@@ -0,0 +1,25 @@
+using Newtonsoft.Json.Linq;
+using Papst.EventStore.Documents;
+
+namespace Papst.EventStore;
+
+public interface ILowLevelEventStream
+{
+  /// <summary>
+  /// Appends an Low Level Event to the Stream.
+  /// The Event is encoded as a JObject and the Event Type is provided as a string. This allows to store Events that are not known at compile time.
+  /// </summary>
+  /// <param name="id"></param>
+  /// <param name="eventType"></param>
+  /// <param name="evt"></param>
+  /// <param name="metaData"></param>
+  /// <param name="cancellationToken"></param>
+  /// <returns></returns>
+  Task AppendAsync(
+    Guid id,
+    string eventType,
+    JObject evt,
+    EventStreamMetaData? metaData = null,
+    CancellationToken cancellationToken = default
+  );
+}

tests/Papst.EventStore.AzureCosmos.Tests/CosmosDbIntegrationTestFixture.cs

@@ -14,16 +14,14 @@ public class CosmosDbIntegrationTestFixture : IAsyncLifetime
   public string ContainerName => CosmosContainerId;
 
   private const ushort InternalPort = 8081;
+  //private const string CosmosEmulatorImage = "mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-preview-testcontainers";
+  private const string CosmosEmulatorImage = "mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-EN20260430pre";
 
   private readonly CosmosDbContainer _cosmosDbContainer =
-    //new CosmosDbBuilder("mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-preview")
-    new CosmosDbBuilder("mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-preview")
+    new CosmosDbBuilder(CosmosEmulatorImage)
       .WithPortBinding(InternalPort, true)
-      //.WithEnvironment("AZURE_COSMOS_EMULATOR_PARTITION_COUNT", "10")
       .WithEnvironment("AZURE_COSMOS_EMULATOR_ENABLE_DATA_PERSISTANCE", "false")
-      //.WithCommand("--protocol", "https")
       .WithAutoRemove(true)
-      //.WithWaitStrategy(Wait.ForUnixContainer().AddCustomWaitStrategy(new WaitUntil()))
       .Build();
   private CosmosClient? _cosmosClient;