From 178bc63cf08d6d56c43bc8791da183b20378eed2 Mon Sep 17 00:00:00 2001 From: Marco Papst Date: Thu, 14 May 2026 23:09:35 +0200 Subject: [PATCH 1/2] Update packages: MongoDB.Driver, add SharpCompress Reformat Directory.Packages.props for readability. Bump MongoDB.Driver from 3.7.1 to 3.8.1. Add SharpCompress 0.48.1 to fix a vulnerability. --- Directory.Packages.props | 113 ++++++++++++++++++++------------------- 1 file changed, 57 insertions(+), 56 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 45fd41b..458913c 100755 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -1,58 +1,59 @@ - - true - true - - - - - - - all - runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - all - runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - - - - all - runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - - - - - - - - - - - - - - - - - - - - - - - all - runtime; build; native; contentfiles; analyzers; buildtransitive - - + + true + true + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + + + + + + + + + + + + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + From 6bbdd2f958a5cfd0242088a86b7a8e3004c135c0 Mon Sep 17 00:00:00 2001 From: Marco Papst Date: Sat, 16 May 2026 16:50:44 +0200 Subject: [PATCH 2/2] Add ILowLevelEventStream and Cosmos low-level append API Introduce ILowLevelEventStream for raw event appending with JObject payloads and explicit event type names. Implement this interface in CosmosEventStream, refactor append logic for reuse, and update snapshot handling. Add integration test for low-level append. Update Cosmos emulator image and document the new API in README. --- README.md | 32 +++++ .../CosmosEventStream.cs | 136 +++++++++--------- src/Papst.EventStore/ILowLevelEventStream.cs | 25 ++++ .../CosmosDbIntegrationTestFixture.cs | 8 +- .../CosmosEventStreamTests.cs | 25 ++++ 5 files changed, 154 insertions(+), 72 deletions(-) create mode 100644 src/Papst.EventStore/ILowLevelEventStream.cs diff --git a/README.md b/README.md index e886a62..83c4675 100644 --- a/README.md +++ b/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()`. + ## 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. diff --git a/src/Papst.EventStore.AzureCosmos/CosmosEventStream.cs b/src/Papst.EventStore.AzureCosmos/CosmosEventStream.cs index 4c42e14..5c3a19f 100755 --- a/src/Papst.EventStore.AzureCosmos/CosmosEventStream.cs +++ b/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( Guid id, TEvent evt, @@ -90,52 +110,7 @@ public async Task AppendAsync( { 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 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 indexPatch = await dbProvider.Container - .PatchItemAsync( - _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( @@ -149,31 +124,37 @@ public async Task AppendSnapshotAsync( 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(_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 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 patches = CreateAppendPatches(metaData, additionalPatches); ItemResponse indexPatch = await dbProvider.Container .PatchItemAsync( _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( 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(_stream.Id, new(StreamId.ToString()), cancellationToken: cancellationToken) - .ConfigureAwait(false); + private List CreateAppendPatches( + EventStreamMetaData? metaData, + params PatchOperation[] additionalPatches + ) + { + List 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 CreateEventEntity( Guid id, @@ -532,7 +533,7 @@ private async IAsyncEnumerable CreateBatches(IReadOnlyList CreateBatches(IReadOnlyList + /// 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. + /// + /// + /// + /// + /// + /// + /// + Task AppendAsync( + Guid id, + string eventType, + JObject evt, + EventStreamMetaData? metaData = null, + CancellationToken cancellationToken = default + ); +} diff --git a/tests/Papst.EventStore.AzureCosmos.Tests/CosmosDbIntegrationTestFixture.cs b/tests/Papst.EventStore.AzureCosmos.Tests/CosmosDbIntegrationTestFixture.cs index 97ab911..63c4399 100755 --- a/tests/Papst.EventStore.AzureCosmos.Tests/CosmosDbIntegrationTestFixture.cs +++ b/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; diff --git a/tests/Papst.EventStore.AzureCosmos.Tests/IntegrationTests/CosmosEventStreamTests.cs b/tests/Papst.EventStore.AzureCosmos.Tests/IntegrationTests/CosmosEventStreamTests.cs index cf2a69d..7ff760d 100755 --- a/tests/Papst.EventStore.AzureCosmos.Tests/IntegrationTests/CosmosEventStreamTests.cs +++ b/tests/Papst.EventStore.AzureCosmos.Tests/IntegrationTests/CosmosEventStreamTests.cs @@ -1,9 +1,11 @@ using AutoFixture.Xunit2; using Microsoft.Extensions.DependencyInjection; +using Newtonsoft.Json.Linq; using Papst.EventStore.Aggregation; using Papst.EventStore.AzureCosmos.Tests.IntegrationTests.Models; using Papst.EventStore.Documents; using Shouldly; +using System.Linq; using Xunit; namespace Papst.EventStore.AzureCosmos.Tests.IntegrationTests; @@ -65,6 +67,29 @@ public async Task AppendAsync_ShouldAppendEvent(Guid streamId, Guid documentId) events.ShouldContain(d => d.Id == documentId); } + [Theory, AutoData] + public async Task LowLevelAppendAsync_ShouldAppendEvent(Guid streamId, Guid documentId) + { + var services = _fixture.BuildServiceProvider(); + var store = services.GetRequiredService(); + var stream = await CreateStreamAsync(store, streamId); + var lowLevelStream = stream.ShouldBeAssignableTo(); + JObject payload = new() + { + ["value"] = "low-level" + }; + + await lowLevelStream.AppendAsync(documentId, "LowLevelEvent", payload); + + stream.Version.ShouldBe(1UL); + var events = await stream.ListAsync().ToListAsync(); + events.ShouldContain(d => d.Id == documentId); + EventStreamDocument appendedEvent = events.Single(d => d.Id == documentId); + appendedEvent.DataType.ShouldBe("LowLevelEvent"); + appendedEvent.Name.ShouldBe("LowLevelEvent"); + appendedEvent.Data.ShouldBe(payload); + } + [Theory, AutoData] public async Task AppendTransactionAsync_ShouldAppendEvents(Guid streamId) {