diff --git a/Directory.Packages.props b/Directory.Packages.props
index 0826e13e4e..37aabf94fd 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -172,13 +172,20 @@
half (a timed-out blue/green side-effect gate re-reads progression and succeeds when the replay
had already reached the mark, plus a configurable DaemonSettings.SideEffectGateTimeout),
jasperfx#595 (BatchingChannel could deliver its trailing batch twice on shutdown) and #597. -->
-
-
-
+
+
+
+
+
all
runtime; build; native; contentfiles; analyzers; buildtransitive
-
+
diff --git a/docs/events/dcb.md b/docs/events/dcb.md
index f5e8c08b12..5a04731c25 100644
--- a/docs/events/dcb.md
+++ b/docs/events/dcb.md
@@ -37,7 +37,7 @@ private void ConfigureStore()
});
}
```
-snippet source | anchor
+snippet source | anchor
Each tag type gets its own table (`mt_event_tag_student`, `mt_event_tag_course`, etc.) with a composite primary key of `(value, seq_id)`.
@@ -75,7 +75,7 @@ Tag types should be simple wrapper records around a primitive value:
public record StudentId(Guid Value);
public record CourseId(Guid Value);
```
-snippet source | anchor
+snippet source | anchor
Supported inner value types: `Guid`, `string`, `int`, `long`, `short`.
@@ -94,7 +94,7 @@ enrolled.WithTag(studentId, courseId);
theSession.Events.Append(streamId, enrolled);
await theSession.SaveChangesAsync();
```
-snippet source | anchor
+snippet source | anchor
Events can have multiple tags of different types. Tags are persisted to their respective tag tables in the same transaction as the event.
@@ -109,7 +109,7 @@ Use `EventTagQuery` to build a query, then execute it with `QueryByTagsAsync`:
var query = new EventTagQuery().Or(studentId);
var events = await theSession.Events.QueryByTagsAsync(query);
```
-snippet source | anchor
+snippet source | anchor
### Multiple Tags (OR)
@@ -124,7 +124,7 @@ var query = new EventTagQuery()
var events = await theSession.Events.QueryByTagsAsync(query);
```
-snippet source | anchor
+snippet source | anchor
### Filtering by Event Type
@@ -138,7 +138,7 @@ var query = new EventTagQuery()
var events = await theSession.Events.QueryByTagsAsync(query);
```
-snippet source | anchor
+snippet source | anchor
Events are always returned ordered by sequence number (global append order).
@@ -176,7 +176,7 @@ public class StudentCourseEnrollment
}
}
```
-snippet source | anchor
+snippet source | anchor
Then aggregate across streams by tag query:
@@ -190,7 +190,7 @@ var query = new EventTagQuery()
var aggregate = await theSession.Events.AggregateByTagsAsync(query);
```
-snippet source | anchor
+snippet source | anchor
Returns `null` if no matching events are found.
@@ -250,7 +250,7 @@ boundary.AppendOne(assignment);
// appended matching events after our read
await session2.SaveChangesAsync();
```
-snippet source | anchor
+snippet source | anchor
### Handling Concurrency Violations
@@ -269,7 +269,7 @@ catch (DcbConcurrencyException ex)
// ex.LastSeenSequence -- the sequence at time of read
}
```
-snippet source | anchor
+snippet source | anchor
::: tip
@@ -313,7 +313,7 @@ public async Task events_exist_returns_true_when_matching_events_found()
exists.ShouldBeTrue();
}
```
-snippet source | anchor
+snippet source | anchor
This is useful for guard clauses and validation logic in DCB workflows where you need to check preconditions before appending new events.
diff --git a/src/EventSourcingTests/Aggregation/auto_discover_aggregate_types.cs b/src/EventSourcingTests/Aggregation/auto_discover_aggregate_types.cs
index 7d35841921..2fea327da3 100644
--- a/src/EventSourcingTests/Aggregation/auto_discover_aggregate_types.cs
+++ b/src/EventSourcingTests/Aggregation/auto_discover_aggregate_types.cs
@@ -2,6 +2,7 @@
using System.Linq;
using System.Threading.Tasks;
using JasperFx.Events;
+using JasperFx.Events.ComplianceTests;
using Marten;
using Marten.Testing.Harness;
using Shouldly;
@@ -36,7 +37,7 @@ public async Task auto_discovered_type_works_for_live_aggregation()
{
// No explicit Snapshot() registration — relies on auto-discovery
var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new CEvent());
+ theSession.Events.StartStream(streamId, new EvolveAEvent(), new EvolveBEvent(), new EvolveCEvent());
await theSession.SaveChangesAsync();
var aggregate = await theSession.Events.AggregateStreamAsync(streamId);
diff --git a/src/EventSourcingTests/Aggregation/self_aggregating_evolve_method.cs b/src/EventSourcingTests/Aggregation/self_aggregating_evolve_method.cs
deleted file mode 100644
index d559adbb1a..0000000000
--- a/src/EventSourcingTests/Aggregation/self_aggregating_evolve_method.cs
+++ /dev/null
@@ -1,319 +0,0 @@
-using System;
-using System.Threading.Tasks;
-using JasperFx.Events;
-using Marten;
-using Marten.Events;
-using Marten.Events.Projections;
-using Marten.Testing.Harness;
-using Shouldly;
-using Xunit;
-
-namespace EventSourcingTests.Aggregation;
-
-#region sample_evolve_aggregates
-
-///
-/// Mutable aggregate using void Evolve(IEvent e) — switches on IEvent envelope
-///
-public class MutableIEventEvolveAggregate
-{
- public Guid Id { get; set; }
- public int ACount { get; set; }
- public int BCount { get; set; }
- public int CCount { get; set; }
-
- public void Evolve(IEvent e)
- {
- switch (e)
- {
- case IEvent:
- ACount++;
- break;
- case IEvent:
- BCount++;
- break;
- case IEvent:
- CCount++;
- break;
- }
- }
-}
-
-///
-/// Mutable aggregate using void Evolve(object o) — switches on event data
-///
-public class MutableObjectEvolveAggregate
-{
- public Guid Id { get; set; }
- public int ACount { get; set; }
- public int BCount { get; set; }
- public int CCount { get; set; }
-
- public void Evolve(object o)
- {
- switch (o)
- {
- case AEvent:
- ACount++;
- break;
- case BEvent:
- BCount++;
- break;
- case CEvent:
- CCount++;
- break;
- }
- }
-}
-
-///
-/// Immutable aggregate using TDoc Evolve(IEvent e) — returns new instance
-///
-public record ImmutableIEventEvolveAggregate(Guid Id, int ACount = 0, int BCount = 0, int CCount = 0)
-{
- public ImmutableIEventEvolveAggregate() : this(Guid.Empty) { }
-
- public ImmutableIEventEvolveAggregate Evolve(IEvent e)
- {
- return e switch
- {
- IEvent => this with { ACount = ACount + 1 },
- IEvent => this with { BCount = BCount + 1 },
- IEvent => this with { CCount = CCount + 1 },
- _ => this
- };
- }
-}
-
-///
-/// Immutable aggregate using TDoc Evolve(object o) — returns new instance
-///
-public record ImmutableObjectEvolveAggregate(Guid Id, int ACount = 0, int BCount = 0, int CCount = 0)
-{
- public ImmutableObjectEvolveAggregate() : this(Guid.Empty) { }
-
- public ImmutableObjectEvolveAggregate Evolve(object o)
- {
- return o switch
- {
- AEvent => this with { ACount = ACount + 1 },
- BEvent => this with { BCount = BCount + 1 },
- CEvent => this with { CCount = CCount + 1 },
- _ => this
- };
- }
-}
-
-///
-/// Mutable aggregate using async Task EvolveAsync(IEvent e, IQuerySession session)
-///
-public class AsyncEvolveAggregate
-{
- public Guid Id { get; set; }
- public int ACount { get; set; }
- public int BCount { get; set; }
-
- public Task EvolveAsync(IEvent e, IQuerySession session)
- {
- switch (e)
- {
- case IEvent:
- ACount++;
- break;
- case IEvent:
- BCount++;
- break;
- }
-
- return Task.CompletedTask;
- }
-}
-
-///
-/// Immutable aggregate using async ValueTask<TDoc> EvolveAsync(IEvent e, IQuerySession session)
-///
-public record ImmutableAsyncEvolveAggregate(Guid Id, int ACount = 0, int BCount = 0)
-{
- public ImmutableAsyncEvolveAggregate() : this(Guid.Empty) { }
-
- public ValueTask EvolveAsync(IEvent e, IQuerySession session)
- {
- var result = e switch
- {
- IEvent => this with { ACount = ACount + 1 },
- IEvent => this with { BCount = BCount + 1 },
- _ => this
- };
-
- return new ValueTask(result);
- }
-}
-
-#endregion
-
-///
-/// Tests for self-aggregating types that use Evolve/EvolveAsync methods
-/// instead of conventional Apply/Create methods. The source generator
-/// creates IGeneratedSyncEvolver or IGeneratedAsyncEvolver implementations
-/// that delegate to the user's Evolve/EvolveAsync method.
-///
-public class self_aggregating_evolve_method : IntegrationContext
-{
- public self_aggregating_evolve_method(DefaultStoreFixture fixture) : base(fixture)
- {
- }
-
- [Fact]
- public async Task mutable_ievent_evolve_inline()
- {
- StoreOptions(opts =>
- {
- opts.Projections.Snapshot(SnapshotLifecycle.Inline);
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new AEvent(), new CEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.LoadAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(2);
- aggregate.BCount.ShouldBe(1);
- aggregate.CCount.ShouldBe(1);
- }
-
- [Fact]
- public async Task mutable_object_evolve_inline()
- {
- StoreOptions(opts =>
- {
- opts.Projections.Snapshot(SnapshotLifecycle.Inline);
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new CEvent(), new CEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.LoadAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(1);
- aggregate.BCount.ShouldBe(1);
- aggregate.CCount.ShouldBe(2);
- }
-
- [Fact]
- public async Task immutable_ievent_evolve_inline()
- {
- StoreOptions(opts =>
- {
- opts.Projections.Snapshot(SnapshotLifecycle.Inline);
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new AEvent(), new BEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.LoadAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(2);
- aggregate.BCount.ShouldBe(1);
- aggregate.CCount.ShouldBe(0);
- }
-
- [Fact]
- public async Task immutable_object_evolve_inline()
- {
- StoreOptions(opts =>
- {
- opts.Projections.Snapshot(SnapshotLifecycle.Inline);
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new BEvent(), new CEvent(), new AEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.LoadAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(1);
- aggregate.BCount.ShouldBe(1);
- aggregate.CCount.ShouldBe(1);
- }
-
- [Fact]
- public async Task async_evolve_inline()
- {
- StoreOptions(opts =>
- {
- opts.Projections.Snapshot(SnapshotLifecycle.Inline);
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new AEvent(), new BEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.LoadAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(2);
- aggregate.BCount.ShouldBe(1);
- }
-
- [Fact]
- public async Task immutable_async_evolve_inline()
- {
- StoreOptions(opts =>
- {
- opts.Projections.Snapshot(SnapshotLifecycle.Inline);
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new AEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.LoadAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(2);
- aggregate.BCount.ShouldBe(1);
- }
-
- [Fact]
- public async Task mutable_ievent_evolve_with_append_to_existing_stream()
- {
- StoreOptions(opts =>
- {
- opts.Projections.Snapshot(SnapshotLifecycle.Inline);
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new BEvent());
- await theSession.SaveChangesAsync();
-
- // Append more events
- theSession.Events.Append(streamId, new AEvent(), new CEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.LoadAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(2);
- aggregate.BCount.ShouldBe(1);
- aggregate.CCount.ShouldBe(1);
- }
-
- [Fact]
- public async Task live_aggregation_with_evolve()
- {
- StoreOptions(opts =>
- {
- // No snapshot — live aggregation only
- });
-
- var streamId = Guid.NewGuid();
- theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new CEvent());
- await theSession.SaveChangesAsync();
-
- var aggregate = await theSession.Events.AggregateStreamAsync(streamId);
- aggregate.ShouldNotBeNull();
- aggregate.ACount.ShouldBe(1);
- aggregate.BCount.ShouldBe(1);
- aggregate.CCount.ShouldBe(1);
- }
-}
diff --git a/src/EventSourcingTests/Compliance/marten_event_store_compliance.cs b/src/EventSourcingTests/Compliance/marten_event_store_compliance.cs
new file mode 100644
index 0000000000..67aa4d9f49
--- /dev/null
+++ b/src/EventSourcingTests/Compliance/marten_event_store_compliance.cs
@@ -0,0 +1,24 @@
+using JasperFx.Events.ComplianceTests;
+using Marten;
+using Marten.Testing.Harness;
+
+namespace EventSourcingTests.Compliance;
+
+/*
+ * Marten's enrollment in the cross-store event sourcing compliance suites. Each class below is
+ * empty on purpose: the behavior lives once in JasperFx.Events.ComplianceTests and is closed here
+ * over Marten's IEventStore session pair through
+ * MartenComplianceFixture. Polecat -- and later the Sqlite minimal store -- enroll the same way.
+ */
+
+public class self_aggregating_evolve_compliance
+ : SelfAggregatingEvolveCompliance;
+
+public class dcb_tag_query_and_consistency_compliance
+ : DcbTagQueryAndConsistencyCompliance;
+
+public class assign_tag_where_compliance
+ : AssignTagWhereCompliance;
+
+public class async_daemon_compliance
+ : AsyncDaemonCompliance;
diff --git a/src/EventSourcingTests/Dcb/OrderTagTypes.cs b/src/EventSourcingTests/Dcb/OrderTagTypes.cs
new file mode 100644
index 0000000000..efe2d92e7a
--- /dev/null
+++ b/src/EventSourcingTests/Dcb/OrderTagTypes.cs
@@ -0,0 +1,19 @@
+#nullable enable
+using System;
+
+namespace EventSourcingTests.Dcb;
+
+///
+/// The retroactive-tagging fixture types. The portable AssignTagWhere behavior now lives in
+/// ;
+/// these records stay here because Marten's HStore-specific
+/// exercises the same domain against a storage mode
+/// Marten alone supports.
+///
+public record RegionId(Guid Value);
+
+public record OrderPlaced(string OrderNumber, decimal Amount);
+
+public record OrderShipped(string OrderNumber);
+
+public record OrderCancelled(string OrderNumber, string Reason);
diff --git a/src/EventSourcingTests/Dcb/assign_tag_where_tests.cs b/src/EventSourcingTests/Dcb/assign_tag_where_tests.cs
deleted file mode 100644
index 3fffa30df4..0000000000
--- a/src/EventSourcingTests/Dcb/assign_tag_where_tests.cs
+++ /dev/null
@@ -1,207 +0,0 @@
-#nullable enable
-using System;
-using System.Linq;
-using System.Threading.Tasks;
-using JasperFx.Events;
-using JasperFx.Events.Tags;
-using Marten;
-using Marten.Events;
-using Marten.Testing.Harness;
-using Shouldly;
-using Xunit;
-
-namespace EventSourcingTests.Dcb;
-
-public record RegionId(Guid Value);
-
-public record OrderPlaced(string OrderNumber, decimal Amount);
-public record OrderShipped(string OrderNumber);
-public record OrderCancelled(string OrderNumber, string Reason);
-
-[Collection("OneOffs")]
-public class assign_tag_where_tests : OneOffConfigurationsContext, IAsyncLifetime
-{
- private RegionId _eastRegion = null!;
- private RegionId _westRegion = null!;
-
- public override ValueTask InitializeAsync()
- {
- _eastRegion = new RegionId(Guid.NewGuid());
- _westRegion = new RegionId(Guid.NewGuid());
-
- StoreOptions(opts =>
- {
- opts.Events.AddEventType();
- opts.Events.AddEventType();
- opts.Events.AddEventType();
-
- opts.Events.RegisterTagType("region");
- });
-
- return default;
- }
-
- public override ValueTask DisposeAsync() => base.DisposeAsync();
-
- [Fact]
- public async Task assign_tag_where_by_event_type_name()
- {
- // Append events WITHOUT tags
- var stream1 = Guid.NewGuid();
- theSession.Events.Append(stream1,
- new OrderPlaced("ORD-1", 100m),
- new OrderShipped("ORD-1"));
- await theSession.SaveChangesAsync();
-
- // Now retroactively tag all OrderPlaced events with a region
- await using var session2 = theStore.LightweightSession();
- var orderPlacedTypeName = theStore.Options.EventGraph.EventMappingFor().EventTypeName;
- session2.Events.AssignTagWhere(
- e => e.EventTypeName == orderPlacedTypeName,
- _eastRegion);
- await session2.SaveChangesAsync();
-
- // Query by tag - should find only the OrderPlaced event
- await using var session3 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(_eastRegion);
- var events = await session3.Events.QueryByTagsAsync(query);
-
- events.Count.ShouldBe(1);
- events[0].Data.ShouldBeOfType().OrderNumber.ShouldBe("ORD-1");
- }
-
- [Fact]
- public async Task assign_tag_where_by_stream_id()
- {
- var stream1 = Guid.NewGuid();
- var stream2 = Guid.NewGuid();
-
- theSession.Events.Append(stream1,
- new OrderPlaced("ORD-1", 100m),
- new OrderShipped("ORD-1"));
- theSession.Events.Append(stream2,
- new OrderPlaced("ORD-2", 200m));
- await theSession.SaveChangesAsync();
-
- // Tag all events in stream1 only
- await using var session2 = theStore.LightweightSession();
- session2.Events.AssignTagWhere(
- e => e.StreamId == stream1,
- _eastRegion);
- await session2.SaveChangesAsync();
-
- // Query - should find only the 2 events from stream1
- await using var session3 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(_eastRegion);
- var events = await session3.Events.QueryByTagsAsync(query);
-
- events.Count.ShouldBe(2);
- events.ShouldAllBe(e => e.StreamId == stream1);
- }
-
- [Fact]
- public async Task assign_tag_where_with_compound_predicate()
- {
- var stream1 = Guid.NewGuid();
-
- theSession.Events.Append(stream1,
- new OrderPlaced("ORD-1", 100m),
- new OrderShipped("ORD-1"),
- new OrderCancelled("ORD-1", "changed mind"));
- await theSession.SaveChangesAsync();
-
- // Tag events that are of type OrderPlaced or OrderCancelled
- await using var session2 = theStore.LightweightSession();
- var placedType = theStore.Options.EventGraph.EventMappingFor().EventTypeName;
- var cancelledType = theStore.Options.EventGraph.EventMappingFor().EventTypeName;
-
- session2.Events.AssignTagWhere(
- e => e.EventTypeName == placedType || e.EventTypeName == cancelledType,
- _eastRegion);
- await session2.SaveChangesAsync();
-
- // Query - should find 2 events (placed + cancelled, NOT shipped)
- await using var session3 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(_eastRegion);
- var events = await session3.Events.QueryByTagsAsync(query);
-
- events.Count.ShouldBe(2);
- events.Select(e => e.Data.GetType()).ShouldContain(typeof(OrderPlaced));
- events.Select(e => e.Data.GetType()).ShouldContain(typeof(OrderCancelled));
- events.Select(e => e.Data.GetType()).ShouldNotContain(typeof(OrderShipped));
- }
-
- [Fact]
- public async Task assign_tag_where_is_idempotent()
- {
- var stream1 = Guid.NewGuid();
- theSession.Events.Append(stream1, new OrderPlaced("ORD-1", 100m));
- await theSession.SaveChangesAsync();
-
- var placedType = theStore.Options.EventGraph.EventMappingFor().EventTypeName;
-
- // Assign the same tag twice - should not fail or duplicate
- await using var session2 = theStore.LightweightSession();
- session2.Events.AssignTagWhere(
- e => e.EventTypeName == placedType, _eastRegion);
- await session2.SaveChangesAsync();
-
- await using var session3 = theStore.LightweightSession();
- session3.Events.AssignTagWhere(
- e => e.EventTypeName == placedType, _eastRegion);
- await session3.SaveChangesAsync();
-
- // Should still just find 1 event
- await using var session4 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(_eastRegion);
- var events = await session4.Events.QueryByTagsAsync(query);
- events.Count.ShouldBe(1);
- }
-
- [Fact]
- public async Task assign_tag_where_does_not_affect_unmatched_events()
- {
- var stream1 = Guid.NewGuid();
- var stream2 = Guid.NewGuid();
-
- theSession.Events.Append(stream1, new OrderPlaced("ORD-1", 100m));
- theSession.Events.Append(stream2, new OrderPlaced("ORD-2", 200m));
- await theSession.SaveChangesAsync();
-
- // Only tag events in stream1
- await using var session2 = theStore.LightweightSession();
- session2.Events.AssignTagWhere(
- e => e.StreamId == stream1, _eastRegion);
- await session2.SaveChangesAsync();
-
- // Tag events in stream2 with different region
- await using var session3 = theStore.LightweightSession();
- session3.Events.AssignTagWhere(
- e => e.StreamId == stream2, _westRegion);
- await session3.SaveChangesAsync();
-
- // Verify east only has stream1
- await using var session4 = theStore.LightweightSession();
- var eastEvents = await session4.Events.QueryByTagsAsync(
- new EventTagQuery().Or(_eastRegion));
- eastEvents.Count.ShouldBe(1);
- eastEvents[0].StreamId.ShouldBe(stream1);
-
- // Verify west only has stream2
- var westEvents = await session4.Events.QueryByTagsAsync(
- new EventTagQuery().Or(_westRegion));
- westEvents.Count.ShouldBe(1);
- westEvents[0].StreamId.ShouldBe(stream2);
- }
-
- [Fact]
- public async Task assign_tag_where_throws_for_unregistered_tag_type()
- {
- var unregisteredTag = new StudentId(Guid.NewGuid());
-
- Should.Throw(() =>
- {
- theSession.Events.AssignTagWhere(e => e.Sequence > 0, unregisteredTag);
- });
- }
-}
diff --git a/src/EventSourcingTests/Dcb/dcb_documentation_samples.cs b/src/EventSourcingTests/Dcb/dcb_documentation_samples.cs
new file mode 100644
index 0000000000..37def82a5a
--- /dev/null
+++ b/src/EventSourcingTests/Dcb/dcb_documentation_samples.cs
@@ -0,0 +1,314 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using JasperFx.Events;
+using JasperFx.Events.Tags;
+using Marten;
+using Marten.Events;
+using Marten.Events.Dcb;
+using Marten.Testing.Harness;
+using Shouldly;
+using Xunit;
+
+namespace EventSourcingTests.Dcb;
+
+#region sample_marten_dcb_tag_type_definitions
+// Strong-typed tag identifiers
+public record StudentId(Guid Value);
+public record CourseId(Guid Value);
+#endregion
+
+#region sample_marten_dcb_domain_events
+// Domain events
+public record StudentEnrolled(string StudentName, string CourseName);
+public record AssignmentSubmitted(string AssignmentName, int Score);
+public record StudentDropped(string Reason);
+#endregion
+
+// Event with tag-typed properties for inference testing
+public record StudentGraded(StudentId StudentId, CourseId CourseId, int Grade);
+
+// Event with NO tag-typed properties — should fail inference
+public record SystemNotification(string Message);
+
+#region sample_marten_dcb_aggregate
+// Aggregate for DCB
+public class StudentCourseEnrollment
+{
+ public Guid Id { get; set; }
+ public string StudentName { get; set; } = "";
+ public string CourseName { get; set; } = "";
+ public List Assignments { get; set; } = new();
+ public bool IsDropped { get; set; }
+
+ public void Apply(StudentEnrolled e)
+ {
+ StudentName = e.StudentName;
+ CourseName = e.CourseName;
+ }
+
+ public void Apply(AssignmentSubmitted e)
+ {
+ Assignments.Add(e.AssignmentName);
+ }
+
+ public void Apply(StudentDropped e)
+ {
+ IsDropped = true;
+ }
+}
+#endregion
+
+///
+/// The executable source of the DCB documentation samples in docs/events/dcb.md, and the
+/// home of the tag/event/aggregate types the other Marten-specific DCB test fixtures share.
+///
+///
+/// The behavioral coverage that used to live here now runs once in
+///
+/// against every Critter Stack event store. What stays behind is the documentation: each test below
+/// backs a sample_marten_dcb_* snippet block, so it has to keep compiling and passing with
+/// Marten-flavored API calls in it.
+///
+[Collection("OneOffs")]
+public class dcb_documentation_samples: OneOffConfigurationsContext, IAsyncLifetime
+{
+ #region sample_marten_dcb_registering_tag_types
+ private void ConfigureStore()
+ {
+ StoreOptions(opts =>
+ {
+ opts.Events.AddEventType();
+ opts.Events.AddEventType();
+ opts.Events.AddEventType();
+ opts.Events.AddEventType();
+
+ // Register tag types -- each gets its own table (mt_event_tag_student, mt_event_tag_course)
+ opts.Events.RegisterTagType("student")
+ .ForAggregate();
+ opts.Events.RegisterTagType("course")
+ .ForAggregate();
+
+ opts.Projections.LiveStreamAggregation();
+ });
+ }
+ #endregion
+
+ public override ValueTask InitializeAsync()
+ {
+ ConfigureStore();
+ return default;
+ }
+
+ public override ValueTask DisposeAsync() => base.DisposeAsync();
+
+ [Fact]
+ public async Task can_query_events_by_single_tag()
+ {
+ var studentId = new StudentId(Guid.NewGuid());
+ var courseId = new CourseId(Guid.NewGuid());
+ var streamId = Guid.NewGuid();
+
+ #region sample_marten_dcb_tagging_events
+ var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
+ enrolled.WithTag(studentId, courseId);
+ theSession.Events.Append(streamId, enrolled);
+ await theSession.SaveChangesAsync();
+ #endregion
+
+ #region sample_marten_dcb_query_by_single_tag
+ var query = new EventTagQuery().Or(studentId);
+ var events = await theSession.Events.QueryByTagsAsync(query);
+ #endregion
+
+ events.Count.ShouldBe(1);
+ events[0].Data.ShouldBeOfType().StudentName.ShouldBe("Alice");
+ }
+
+ [Fact]
+ public async Task can_query_events_by_multiple_tags_with_or()
+ {
+ var student1 = new StudentId(Guid.NewGuid());
+ var student2 = new StudentId(Guid.NewGuid());
+ var course = new CourseId(Guid.NewGuid());
+
+ var e1 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
+ e1.WithTag(student1, course);
+ theSession.Events.Append(Guid.NewGuid(), e1);
+
+ var e2 = theSession.Events.BuildEvent(new StudentEnrolled("Bob", "Math"));
+ e2.WithTag(student2, course);
+ theSession.Events.Append(Guid.NewGuid(), e2);
+
+ await theSession.SaveChangesAsync();
+
+ #region sample_marten_dcb_query_multiple_tags_or
+ // Query for either student
+ var query = new EventTagQuery()
+ .Or(student1)
+ .Or(student2);
+
+ var events = await theSession.Events.QueryByTagsAsync(query);
+ #endregion
+ events.Count.ShouldBe(2);
+ }
+
+ [Fact]
+ public async Task can_query_events_by_tag_with_event_type_filter()
+ {
+ var studentId = new StudentId(Guid.NewGuid());
+ var courseId = new CourseId(Guid.NewGuid());
+ var streamId = Guid.NewGuid();
+
+ var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
+ enrolled.WithTag(studentId, courseId);
+
+ var submitted = theSession.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
+ submitted.WithTag(studentId, courseId);
+
+ theSession.Events.Append(streamId, enrolled, submitted);
+ await theSession.SaveChangesAsync();
+
+ #region sample_marten_dcb_query_by_event_type
+ // Query only AssignmentSubmitted events for this student
+ var query = new EventTagQuery()
+ .Or(studentId);
+
+ var events = await theSession.Events.QueryByTagsAsync(query);
+ #endregion
+ events.Count.ShouldBe(1);
+ events[0].Data.ShouldBeOfType().AssignmentName.ShouldBe("HW1");
+ }
+
+ [Fact]
+ public async Task can_aggregate_events_by_tags()
+ {
+ var studentId = new StudentId(Guid.NewGuid());
+ var courseId = new CourseId(Guid.NewGuid());
+ var streamId = Guid.NewGuid();
+
+ var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
+ enrolled.WithTag(studentId, courseId);
+
+ var submitted = theSession.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
+ submitted.WithTag(studentId, courseId);
+
+ theSession.Events.Append(streamId, enrolled, submitted);
+ await theSession.SaveChangesAsync();
+
+ #region sample_marten_dcb_aggregate_by_tags
+ var query = new EventTagQuery()
+ .Or(studentId)
+ .Or(courseId);
+
+ var aggregate = await theSession.Events.AggregateByTagsAsync(query);
+ #endregion
+ aggregate.ShouldNotBeNull();
+ aggregate.StudentName.ShouldBe("Alice");
+ aggregate.CourseName.ShouldBe("Math");
+ aggregate.Assignments.ShouldContain("HW1");
+ }
+
+ [Fact]
+ public async Task can_fetch_for_writing_by_tags_happy_path()
+ {
+ var studentId = new StudentId(Guid.NewGuid());
+ var courseId = new CourseId(Guid.NewGuid());
+ var streamId = Guid.NewGuid();
+
+ // Seed initial events
+ var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
+ enrolled.WithTag(studentId, courseId);
+ theSession.Events.Append(streamId, enrolled);
+ await theSession.SaveChangesAsync();
+
+ #region sample_marten_dcb_fetch_for_writing_by_tags
+ // Fetch for writing
+ await using var session2 = theStore.LightweightSession();
+ var query = new EventTagQuery().Or(studentId);
+ var boundary = await session2.Events.FetchForWritingByTags(query);
+
+ // Read current state
+ var aggregate = boundary.Aggregate; // may be null if no events yet
+ var lastSequence = boundary.LastSeenSequence;
+
+ // Append via boundary
+ var assignment = session2.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
+ assignment.WithTag(studentId, courseId);
+ boundary.AppendOne(assignment);
+
+ // Save -- will throw DcbConcurrencyException if another session
+ // appended matching events after our read
+ await session2.SaveChangesAsync();
+ #endregion
+
+ boundary.Aggregate.ShouldNotBeNull();
+ boundary.Aggregate!.StudentName.ShouldBe("Alice");
+ boundary.Events.Count.ShouldBe(1);
+ }
+
+ [Fact]
+ public async Task fetch_for_writing_by_tags_detects_concurrency_violation()
+ {
+ var studentId = new StudentId(Guid.NewGuid());
+ var courseId = new CourseId(Guid.NewGuid());
+ var streamId = Guid.NewGuid();
+
+ // Seed initial events
+ var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
+ enrolled.WithTag(studentId, courseId);
+ theSession.Events.Append(streamId, enrolled);
+ await theSession.SaveChangesAsync();
+
+ // Session 1: fetch for writing
+ await using var session1 = theStore.LightweightSession();
+ var query = new EventTagQuery().Or(studentId);
+ var boundary = await session1.Events.FetchForWritingByTags(query);
+
+ // Session 2: append a conflicting event BEFORE session 1 saves
+ await using var session2 = theStore.LightweightSession();
+ var conflicting = session2.Events.BuildEvent(new AssignmentSubmitted("HW-conflict", 50));
+ conflicting.WithTag(studentId, courseId);
+ session2.Events.Append(streamId, conflicting);
+ await session2.SaveChangesAsync();
+
+ // Session 1: try to save — should throw DcbConcurrencyException
+ var assignment = session1.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
+ assignment.WithTag(studentId, courseId);
+ boundary.AppendOne(assignment);
+
+ #region sample_marten_dcb_handling_concurrency
+ try
+ {
+ await session1.SaveChangesAsync();
+ }
+ catch (DcbConcurrencyException ex)
+ {
+ // Reload and retry -- the boundary's tag query had new matching events
+ // ex.Query -- the original tag query
+ // ex.LastSeenSequence -- the sequence at time of read
+ }
+ #endregion
+ }
+
+ #region sample_marten_dcb_events_exist_async
+ [Fact]
+ public async Task events_exist_returns_true_when_matching_events_found()
+ {
+ var studentId = new StudentId(Guid.NewGuid());
+ var courseId = new CourseId(Guid.NewGuid());
+ var streamId = Guid.NewGuid();
+
+ var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
+ enrolled.WithTag(studentId, courseId);
+ theSession.Events.Append(streamId, enrolled);
+ await theSession.SaveChangesAsync();
+
+ // Check existence -- lightweight, no event loading
+ var query = new EventTagQuery().Or(studentId);
+ var exists = await theSession.Events.EventsExistAsync(query);
+ exists.ShouldBeTrue();
+ }
+ #endregion
+}
diff --git a/src/EventSourcingTests/Dcb/dcb_tag_query_and_consistency_tests.cs b/src/EventSourcingTests/Dcb/dcb_tag_query_and_consistency_tests.cs
deleted file mode 100644
index 5ce0c66887..0000000000
--- a/src/EventSourcingTests/Dcb/dcb_tag_query_and_consistency_tests.cs
+++ /dev/null
@@ -1,737 +0,0 @@
-#nullable enable
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
-using JasperFx.Events;
-using JasperFx.Events.Tags;
-using Marten;
-using Marten.Events;
-using Marten.Events.Dcb;
-using Marten.Services.BatchQuerying;
-using Marten.Testing.Harness;
-using Shouldly;
-using Xunit;
-
-namespace EventSourcingTests.Dcb;
-
-#region sample_marten_dcb_tag_type_definitions
-// Strong-typed tag identifiers
-public record StudentId(Guid Value);
-public record CourseId(Guid Value);
-#endregion
-
-#region sample_marten_dcb_domain_events
-// Domain events
-public record StudentEnrolled(string StudentName, string CourseName);
-public record AssignmentSubmitted(string AssignmentName, int Score);
-public record StudentDropped(string Reason);
-#endregion
-
-// Event with tag-typed properties for inference testing
-public record StudentGraded(StudentId StudentId, CourseId CourseId, int Grade);
-
-// Event with NO tag-typed properties — should fail inference
-public record SystemNotification(string Message);
-
-#region sample_marten_dcb_aggregate
-// Aggregate for DCB
-public class StudentCourseEnrollment
-{
- public Guid Id { get; set; }
- public string StudentName { get; set; } = "";
- public string CourseName { get; set; } = "";
- public List Assignments { get; set; } = new();
- public bool IsDropped { get; set; }
-
- public void Apply(StudentEnrolled e)
- {
- StudentName = e.StudentName;
- CourseName = e.CourseName;
- }
-
- public void Apply(AssignmentSubmitted e)
- {
- Assignments.Add(e.AssignmentName);
- }
-
- public void Apply(StudentDropped e)
- {
- IsDropped = true;
- }
-}
-#endregion
-
-[Collection("OneOffs")]
-public class dcb_tag_query_and_consistency_tests: OneOffConfigurationsContext, IAsyncLifetime
-{
- #region sample_marten_dcb_registering_tag_types
- private void ConfigureStore()
- {
- StoreOptions(opts =>
- {
- opts.Events.AddEventType();
- opts.Events.AddEventType();
- opts.Events.AddEventType();
- opts.Events.AddEventType();
-
- // Register tag types -- each gets its own table (mt_event_tag_student, mt_event_tag_course)
- opts.Events.RegisterTagType("student")
- .ForAggregate();
- opts.Events.RegisterTagType("course")
- .ForAggregate();
-
- opts.Projections.LiveStreamAggregation();
- });
- }
- #endregion
-
- public override ValueTask InitializeAsync()
- {
- ConfigureStore();
- return default;
- }
-
- public override ValueTask DisposeAsync() => base.DisposeAsync();
-
- private async Task AppendTaggedEvent(Guid streamId, object eventData, params object[] tags)
- {
- var wrapped = theSession.Events.BuildEvent(eventData);
- wrapped.WithTag(tags);
- theSession.Events.Append(streamId, wrapped);
- await theSession.SaveChangesAsync();
- }
-
- [Fact]
- public async Task can_query_events_by_single_tag()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- #region sample_marten_dcb_tagging_events
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
- #endregion
-
- #region sample_marten_dcb_query_by_single_tag
- var query = new EventTagQuery().Or(studentId);
- var events = await theSession.Events.QueryByTagsAsync(query);
- #endregion
-
- events.Count.ShouldBe(1);
- events[0].Data.ShouldBeOfType().StudentName.ShouldBe("Alice");
- }
-
- [Fact]
- public async Task can_query_events_by_multiple_tags_with_or()
- {
- var student1 = new StudentId(Guid.NewGuid());
- var student2 = new StudentId(Guid.NewGuid());
- var course = new CourseId(Guid.NewGuid());
- var stream1 = Guid.NewGuid();
- var stream2 = Guid.NewGuid();
-
- // Student 1 enrolled
- var e1 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- e1.WithTag(student1, course);
- theSession.Events.Append(stream1, e1);
-
- // Student 2 enrolled
- var e2 = theSession.Events.BuildEvent(new StudentEnrolled("Bob", "Math"));
- e2.WithTag(student2, course);
- theSession.Events.Append(stream2, e2);
-
- await theSession.SaveChangesAsync();
-
- #region sample_marten_dcb_query_multiple_tags_or
- // Query for either student
- var query = new EventTagQuery()
- .Or(student1)
- .Or(student2);
-
- var events = await theSession.Events.QueryByTagsAsync(query);
- #endregion
- events.Count.ShouldBe(2);
- }
-
- [Fact]
- public async Task can_query_events_by_tag_with_event_type_filter()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
-
- var submitted = theSession.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
- submitted.WithTag(studentId, courseId);
-
- theSession.Events.Append(streamId, enrolled, submitted);
- await theSession.SaveChangesAsync();
-
- #region sample_marten_dcb_query_by_event_type
- // Query only AssignmentSubmitted events for this student
- var query = new EventTagQuery()
- .Or(studentId);
-
- var events = await theSession.Events.QueryByTagsAsync(query);
- #endregion
- events.Count.ShouldBe(1);
- events[0].Data.ShouldBeOfType().AssignmentName.ShouldBe("HW1");
- }
-
- [Fact]
- public async Task query_returns_empty_when_no_matching_tags()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var otherStudentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Query for a different student
- var query = new EventTagQuery().Or(otherStudentId);
- var events = await theSession.Events.QueryByTagsAsync(query);
- events.Count.ShouldBe(0);
- }
-
- [Fact]
- public async Task can_aggregate_events_by_tags()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
-
- var submitted = theSession.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
- submitted.WithTag(studentId, courseId);
-
- theSession.Events.Append(streamId, enrolled, submitted);
- await theSession.SaveChangesAsync();
-
- #region sample_marten_dcb_aggregate_by_tags
- var query = new EventTagQuery()
- .Or(studentId)
- .Or(courseId);
-
- var aggregate = await theSession.Events.AggregateByTagsAsync(query);
- #endregion
- aggregate.ShouldNotBeNull();
- aggregate.StudentName.ShouldBe("Alice");
- aggregate.CourseName.ShouldBe("Math");
- aggregate.Assignments.ShouldContain("HW1");
- }
-
- [Fact]
- public async Task aggregate_by_tags_returns_null_when_no_events()
- {
- var studentId = new StudentId(Guid.NewGuid());
-
- var query = new EventTagQuery().Or(studentId);
- var aggregate = await theSession.Events.AggregateByTagsAsync(query);
- aggregate.ShouldBeNull();
- }
-
- [Fact]
- public async Task can_fetch_for_writing_by_tags_happy_path()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- // Seed initial events
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- #region sample_marten_dcb_fetch_for_writing_by_tags
- // Fetch for writing
- await using var session2 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(studentId);
- var boundary = await session2.Events.FetchForWritingByTags(query);
-
- // Read current state
- var aggregate = boundary.Aggregate; // may be null if no events yet
- var lastSequence = boundary.LastSeenSequence;
-
- // Append via boundary
- var assignment = session2.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
- assignment.WithTag(studentId, courseId);
- boundary.AppendOne(assignment);
-
- // Save -- will throw DcbConcurrencyException if another session
- // appended matching events after our read
- await session2.SaveChangesAsync();
- #endregion
-
- boundary.Aggregate.ShouldNotBeNull();
- boundary.Aggregate!.StudentName.ShouldBe("Alice");
- boundary.Events.Count.ShouldBe(1);
- }
-
- [Fact]
- public async Task fetch_for_writing_by_tags_detects_concurrency_violation()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- // Seed initial events
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Session 1: fetch for writing
- await using var session1 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(studentId);
- var boundary = await session1.Events.FetchForWritingByTags(query);
-
- // Session 2: append a conflicting event BEFORE session 1 saves
- await using var session2 = theStore.LightweightSession();
- var conflicting = session2.Events.BuildEvent(new AssignmentSubmitted("HW-conflict", 50));
- conflicting.WithTag(studentId, courseId);
- session2.Events.Append(streamId, conflicting);
- await session2.SaveChangesAsync();
-
- // Session 1: try to save — should throw DcbConcurrencyException
- var assignment = session1.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
- assignment.WithTag(studentId, courseId);
- boundary.AppendOne(assignment);
-
- #region sample_marten_dcb_handling_concurrency
- try
- {
- await session1.SaveChangesAsync();
- }
- catch (DcbConcurrencyException ex)
- {
- // Reload and retry -- the boundary's tag query had new matching events
- // ex.Query -- the original tag query
- // ex.LastSeenSequence -- the sequence at time of read
- }
- #endregion
- }
-
- [Fact]
- public async Task fetch_for_writing_by_tags_no_violation_when_unrelated_events_appended()
- {
- var student1 = new StudentId(Guid.NewGuid());
- var student2 = new StudentId(Guid.NewGuid());
- var course = new CourseId(Guid.NewGuid());
- var stream1 = Guid.NewGuid();
- var stream2 = Guid.NewGuid();
-
- // Seed student1
- var enrolled1 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled1.WithTag(student1, course);
- theSession.Events.Append(stream1, enrolled1);
- await theSession.SaveChangesAsync();
-
- // Session 1: fetch for writing for student1
- await using var session1 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(student1);
- var boundary = await session1.Events.FetchForWritingByTags(query);
-
- // Session 2: append event for DIFFERENT student — should NOT conflict
- await using var session2 = theStore.LightweightSession();
- var enrolled2 = session2.Events.BuildEvent(new StudentEnrolled("Bob", "Math"));
- enrolled2.WithTag(student2, course);
- session2.Events.Append(stream2, enrolled2);
- await session2.SaveChangesAsync();
-
- // Session 1: save should succeed
- var assignment = session1.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
- assignment.WithTag(student1, course);
- boundary.AppendOne(assignment);
-
- await session1.SaveChangesAsync(); // Should not throw
- }
-
- [Fact]
- public async Task events_across_multiple_streams_can_be_queried_by_tag()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var course1 = new CourseId(Guid.NewGuid());
- var course2 = new CourseId(Guid.NewGuid());
- var stream1 = Guid.NewGuid();
- var stream2 = Guid.NewGuid();
-
- // Student enrolled in two courses (different streams)
- var enrolled1 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled1.WithTag(studentId, course1);
- theSession.Events.Append(stream1, enrolled1);
-
- var enrolled2 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Science"));
- enrolled2.WithTag(studentId, course2);
- theSession.Events.Append(stream2, enrolled2);
-
- await theSession.SaveChangesAsync();
-
- // Query all events for this student across streams
- var query = new EventTagQuery().Or(studentId);
- var events = await theSession.Events.QueryByTagsAsync(query);
-
- events.Count.ShouldBe(2);
- }
-
- [Fact]
- public async Task query_events_ordered_by_sequence()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
-
- var hw1 = theSession.Events.BuildEvent(new AssignmentSubmitted("HW1", 90));
- hw1.WithTag(studentId, courseId);
-
- var hw2 = theSession.Events.BuildEvent(new AssignmentSubmitted("HW2", 85));
- hw2.WithTag(studentId, courseId);
-
- theSession.Events.Append(streamId, enrolled, hw1, hw2);
- await theSession.SaveChangesAsync();
-
- var query = new EventTagQuery().Or(studentId);
- var events = await theSession.Events.QueryByTagsAsync(query);
-
- events.Count.ShouldBe(3);
- // Events should be ordered by sequence
- events[0].Sequence.ShouldBeLessThan(events[1].Sequence);
- events[1].Sequence.ShouldBeLessThan(events[2].Sequence);
- }
-
- [Fact]
- public async Task fetch_for_writing_with_empty_result_still_enforces_consistency()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
-
- // Fetch for writing when no events exist
- await using var session1 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(studentId);
- var boundary = await session1.Events.FetchForWritingByTags(query);
-
- boundary.Aggregate.ShouldBeNull();
- boundary.Events.Count.ShouldBe(0);
- boundary.LastSeenSequence.ShouldBe(0);
-
- // Another session appends a matching event before save
- await using var session2 = theStore.LightweightSession();
- var enrolled = session2.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- var streamId = Guid.NewGuid();
- session2.Events.Append(streamId, enrolled);
- await session2.SaveChangesAsync();
-
- // Session 1 tries to save — should detect the new matching event
- var e = session1.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- e.WithTag(studentId, courseId);
- boundary.AppendOne(e);
-
- await Should.ThrowAsync(async () =>
- {
- await session1.SaveChangesAsync();
- });
- }
-
- [Fact]
- public async Task can_fetch_for_writing_by_tags_via_batch_query()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- await using var session2 = theStore.LightweightSession();
- var batch = session2.CreateBatchQuery();
- var query = new EventTagQuery().Or(studentId);
- var boundaryTask = batch.Events.FetchForWritingByTags(query);
- await batch.Execute();
-
- var boundary = await boundaryTask;
- boundary.Aggregate.ShouldNotBeNull();
- boundary.Aggregate!.StudentName.ShouldBe("Alice");
- boundary.Events.Count.ShouldBe(1);
- boundary.LastSeenSequence.ShouldBeGreaterThan(0);
-
- // Append via boundary and save
- var assignment = session2.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
- assignment.WithTag(studentId, courseId);
- boundary.AppendOne(assignment);
- await session2.SaveChangesAsync();
- }
-
- [Fact]
- public async Task batch_query_fetch_for_writing_by_tags_detects_concurrency_violation()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Session 1: fetch via batch query
- await using var session1 = theStore.LightweightSession();
- var batch = session1.CreateBatchQuery();
- var query = new EventTagQuery().Or(studentId);
- var boundaryTask = batch.Events.FetchForWritingByTags(query);
- await batch.Execute();
- var boundary = await boundaryTask;
-
- // Session 2: append conflicting event
- await using var session2 = theStore.LightweightSession();
- var conflicting = session2.Events.BuildEvent(new AssignmentSubmitted("HW-conflict", 50));
- conflicting.WithTag(studentId, courseId);
- session2.Events.Append(streamId, conflicting);
- await session2.SaveChangesAsync();
-
- // Session 1: try to save — should throw
- var assignment = session1.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
- assignment.WithTag(studentId, courseId);
- boundary.AppendOne(assignment);
-
- await Should.ThrowAsync(async () =>
- {
- await session1.SaveChangesAsync();
- });
- }
-
- #region sample_marten_dcb_events_exist_async
- [Fact]
- public async Task events_exist_returns_true_when_matching_events_found()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Check existence -- lightweight, no event loading
- var query = new EventTagQuery().Or(studentId);
- var exists = await theSession.Events.EventsExistAsync(query);
- exists.ShouldBeTrue();
- }
- #endregion
-
- [Fact]
- public async Task events_exist_returns_false_when_no_matching_events()
- {
- var studentId = new StudentId(Guid.NewGuid());
-
- var query = new EventTagQuery().Or(studentId);
- var exists = await theSession.Events.EventsExistAsync(query);
- exists.ShouldBeFalse();
- }
-
- [Fact]
- public async Task events_exist_with_event_type_filter()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Should find StudentEnrolled
- var query1 = new EventTagQuery().Or(studentId);
- (await theSession.Events.EventsExistAsync(query1)).ShouldBeTrue();
-
- // Should NOT find AssignmentSubmitted (none appended)
- var query2 = new EventTagQuery().Or(studentId);
- (await theSession.Events.EventsExistAsync(query2)).ShouldBeFalse();
- }
-
- [Fact]
- public async Task events_exist_via_batch_query_positive()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- await using var session2 = theStore.LightweightSession();
- var batch = session2.CreateBatchQuery();
- var query = new EventTagQuery().Or(studentId);
- var existsTask = batch.Events.EventsExist(query);
- await batch.Execute();
-
- (await existsTask).ShouldBeTrue();
- }
-
- [Fact]
- public async Task events_exist_via_batch_query_negative()
- {
- var studentId = new StudentId(Guid.NewGuid());
-
- await using var session2 = theStore.LightweightSession();
- var batch = session2.CreateBatchQuery();
- var query = new EventTagQuery().Or(studentId);
- var existsTask = batch.Events.EventsExist(query);
- await batch.Execute();
-
- (await existsTask).ShouldBeFalse();
- }
-
- [Fact]
- public async Task fetch_for_writing_by_tags_throws_on_empty_query()
- {
- var query = new EventTagQuery();
- await Should.ThrowAsync(async () =>
- {
- await theSession.Events.FetchForWritingByTags(query);
- });
- }
-
- [Fact]
- public async Task append_event_with_inferred_tags_from_properties()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- // Seed initial event with explicit tags
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Fetch for writing
- await using var session2 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(studentId);
- var boundary = await session2.Events.FetchForWritingByTags(query);
-
- // Append a raw event that has StudentId and CourseId properties —
- // tags should be inferred automatically
- boundary.AppendOne(new StudentGraded(studentId, courseId, 95));
-
- // Should succeed — tags inferred from properties
- await session2.SaveChangesAsync();
-
- // Verify the event is discoverable by tag query
- await using var session3 = theStore.LightweightSession();
- var events = await session3.Events.QueryByTagsAsync(
- new EventTagQuery().Or(studentId));
- events.Count.ShouldBe(2);
- events[1].Data.ShouldBeOfType().Grade.ShouldBe(95);
- }
-
- [Fact]
- public async Task append_event_with_no_tags_and_no_inferable_properties_throws()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- // Seed initial event
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Fetch for writing
- await using var session2 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(studentId);
- var boundary = await session2.Events.FetchForWritingByTags(query);
-
- // Append an event with no tags and no tag-typed properties — should throw
- Should.Throw(() =>
- {
- boundary.AppendOne(new SystemNotification("test"));
- });
- }
-
- [Fact]
- public async Task append_already_wrapped_event_with_explicit_tags_works()
- {
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- // Seed initial event
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- // Fetch for writing
- await using var session2 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(studentId);
- var boundary = await session2.Events.FetchForWritingByTags(query);
-
- // Append an already-wrapped event with explicit tags
- var graded = session2.Events.BuildEvent(new StudentGraded(studentId, courseId, 88));
- graded.WithTag(studentId, courseId);
- boundary.AppendOne(graded);
-
- await session2.SaveChangesAsync();
- }
-
- [Fact]
- public async Task append_event_with_tag_having_no_aggregate_type_creates_new_stream()
- {
- // Register a tag type WITHOUT an aggregate association
- StoreOptions(opts =>
- {
- opts.Events.AddEventType();
- opts.Events.AddEventType();
-
- opts.Events.RegisterTagType("student");
- // CourseId registered WITHOUT ForAggregate
- opts.Events.RegisterTagType("course");
-
- opts.Projections.LiveStreamAggregation();
- });
-
- var studentId = new StudentId(Guid.NewGuid());
- var courseId = new CourseId(Guid.NewGuid());
- var streamId = Guid.NewGuid();
-
- var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
- enrolled.WithTag(studentId, courseId);
- theSession.Events.Append(streamId, enrolled);
- await theSession.SaveChangesAsync();
-
- await using var session2 = theStore.LightweightSession();
- var query = new EventTagQuery().Or(studentId);
- var boundary = await session2.Events.FetchForWritingByTags(query);
-
- // CourseId tag has no AggregateType — should create a new stream per event
- var graded = session2.Events.BuildEvent(new StudentGraded(studentId, courseId, 90));
- graded.WithTag(courseId);
- boundary.AppendOne(graded);
-
- // Should succeed — unrouted tag creates a new stream
- await session2.SaveChangesAsync();
- }
-}
diff --git a/src/EventSourcingTests/Dcb/hstore_assign_tag_where_tests.cs b/src/EventSourcingTests/Dcb/hstore_assign_tag_where_tests.cs
index 3825028417..5905f8b5d5 100644
--- a/src/EventSourcingTests/Dcb/hstore_assign_tag_where_tests.cs
+++ b/src/EventSourcingTests/Dcb/hstore_assign_tag_where_tests.cs
@@ -11,13 +11,15 @@
namespace EventSourcingTests.Dcb;
///
-/// Parallel of for .
+/// The HStore counterpart of the shared
+///
+/// suite, for .
/// The retroactive-tag path uses
/// which emits UPDATE mt_events SET tags = COALESCE(tags, ''::hstore) || hstore(...)
/// against rows matching the user-supplied WHERE clause. The merge is naturally
/// idempotent (re-applying the same key-value yields the same hstore).
/// Reuses , , ,
-/// from assign_tag_where_tests.cs.
+/// from OrderTagTypes.cs.
///
[Collection("OneOffs")]
public class hstore_assign_tag_where_tests: OneOffConfigurationsContext, IAsyncLifetime
diff --git a/src/EventSourcingTests/EventSourcingTests.csproj b/src/EventSourcingTests/EventSourcingTests.csproj
index 5491b53e68..4ee5ac514f 100644
--- a/src/EventSourcingTests/EventSourcingTests.csproj
+++ b/src/EventSourcingTests/EventSourcingTests.csproj
@@ -6,6 +6,9 @@
+
+
@@ -66,6 +69,12 @@
Documents\UserWithInheritedId.cs
+
+ Harness\ComplianceQuerySessionAlias.cs
+
+
+ Harness\MartenComplianceFixture.cs
+
Harness\BugIntegrationContext.cs
diff --git a/src/Marten.Testing/Harness/ComplianceQuerySessionAlias.cs b/src/Marten.Testing/Harness/ComplianceQuerySessionAlias.cs
new file mode 100644
index 0000000000..6a1f34a9c6
--- /dev/null
+++ b/src/Marten.Testing/Harness/ComplianceQuerySessionAlias.cs
@@ -0,0 +1,5 @@
+// The shared compliance suites declare self-aggregating types whose EvolveAsync convention method
+// takes the store's own read session. JasperFx's aggregate source generator resolves the parameter
+// by type name, so a per-consumer global alias lets one shared source file bind to Marten's
+// IQuerySession here and to Polecat's in Polecat.
+global using ComplianceQuerySession = Marten.IQuerySession;
diff --git a/src/Marten.Testing/Harness/MartenComplianceFixture.cs b/src/Marten.Testing/Harness/MartenComplianceFixture.cs
new file mode 100644
index 0000000000..7d479f1b66
--- /dev/null
+++ b/src/Marten.Testing/Harness/MartenComplianceFixture.cs
@@ -0,0 +1,146 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using JasperFx;
+using JasperFx.Events;
+using JasperFx.Events.ComplianceTests;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using JasperFx.Events.Tags;
+using Marten.Events;
+using Marten.Services.BatchQuerying;
+
+namespace Marten.Testing.Harness;
+
+///
+/// Marten's implementation of the cross-store event sourcing compliance seam, closing it over
+/// Marten's IEventStore<IDocumentOperations, IQuerySession> session pair.
+///
+public class MartenComplianceFixture: EventStoreComplianceFixture
+{
+ private readonly List