diff --git a/Directory.Packages.props b/Directory.Packages.props index a9d1fdc..e90a1df 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -62,8 +62,14 @@ 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. Same lockstep rule as above. --> - - + + + + - + @@ -159,7 +165,7 @@ - + diff --git a/src/Polecat.Tests/Compliance/ComplianceQuerySessionAlias.cs b/src/Polecat.Tests/Compliance/ComplianceQuerySessionAlias.cs new file mode 100644 index 0000000..ec12b3d --- /dev/null +++ b/src/Polecat.Tests/Compliance/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 Polecat's +// IQuerySession here and to Marten's in Marten. +global using ComplianceQuerySession = Polecat.IQuerySession; diff --git a/src/Polecat.Tests/Compliance/PolecatComplianceFixture.cs b/src/Polecat.Tests/Compliance/PolecatComplianceFixture.cs new file mode 100644 index 0000000..b9973a4 --- /dev/null +++ b/src/Polecat.Tests/Compliance/PolecatComplianceFixture.cs @@ -0,0 +1,153 @@ +using JasperFx; +using JasperFx.Events; +using JasperFx.Events.ComplianceTests; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using JasperFx.Events.Tags; +using Microsoft.Data.SqlClient; +using Polecat.Batching; + +namespace Polecat.Tests.Compliance; + +/// +/// Polecat's implementation of the cross-store event sourcing compliance seam, closing it over +/// Polecat's IEventStore<IDocumentSession, IQuerySession> session pair. +/// +public class PolecatComplianceFixture : EventStoreComplianceFixture +{ + private readonly List _disposables = new(); + private DocumentStore _store = null!; + + public DocumentStore Store => _store; + + protected override async Task BuildStoreAsync(ComplianceStoreConfig config) + { + var schemaName = (config.SchemaName ?? "compliance").ToLowerInvariant(); + + var options = new StoreOptions + { + ConnectionString = ConnectionSource.ConnectionString, + AutoCreateSchemaObjects = AutoCreate.All, + DatabaseSchemaName = schemaName, + UseNativeJsonType = ConnectionSource.SupportsNativeJson + }; + + config.ApplyTo(new PolecatComplianceRegistrar(options)); + + _store = new DocumentStore(options); + _disposables.Add(_store); + + // Polecat applies schema changes explicitly rather than lazily -- one of the eight + // divergences the compliance seam exists to absorb. + await _store.Database.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); + } + + public override IDocumentSession OpenSession() => _store.LightweightSession(); + + public override Task SaveChangesAsync(IDocumentSession session, CancellationToken token) + => session.SaveChangesAsync(token); + + public override Task LoadDocumentAsync(IQuerySession session, object id, CancellationToken token) + where T : class + => id switch + { + Guid guidId => session.LoadAsync(guidId, token), + int intId => session.LoadAsync(intId, token), + long longId => session.LoadAsync(longId, token), + string stringId => session.LoadAsync(stringId, token), + _ => throw new ArgumentOutOfRangeException(nameof(id), + $"Polecat cannot load documents by an identity of type {id.GetType().FullName}") + }; + + public override IEventStoreOperations EventsFor(IDocumentSession session) => session.Events; + + public override IComplianceBatch CreateBatch(IQuerySession session) + => new PolecatComplianceBatch(session.CreateBatchQuery()); + + public override IEventRegistry Registry => _store.Options.EventGraph; + + public override async Task CleanEventDataAsync() + { + await _store.Advanced.Clean.DeleteAllEventDataAsync().ConfigureAwait(false); + await _store.Advanced.Clean.DeleteAllDocumentsAsync().ConfigureAwait(false); + } + + public override async Task StartDaemonAsync() + { + var daemon = await _store.BuildProjectionDaemonAsync().ConfigureAwait(false); + _disposables.Add(daemon); + + await daemon.StartAllAsync().ConfigureAwait(false); + + return daemon; + } + + public override Task WaitForNonStaleProjectionDataAsync(TimeSpan timeout) + => _store.Database.WaitForNonStaleProjectionDataAsync(timeout); + + /// + /// Polecat derives live aggregators automatically from self-aggregating types; there is no + /// explicit registration call to make. + /// + public override bool SupportsLiveAggregationRegistration => false; + + public override async ValueTask DisposeAsync() + { + foreach (var disposable in _disposables) + { + switch (disposable) + { + case IAsyncDisposable asyncDisposable: + await asyncDisposable.DisposeAsync().ConfigureAwait(false); + break; + case IDisposable syncDisposable: + syncDisposable.Dispose(); + break; + } + } + + _disposables.Clear(); + } + + internal class PolecatComplianceRegistrar : IComplianceStoreRegistrar + { + private readonly StoreOptions _options; + + public PolecatComplianceRegistrar(StoreOptions options) + { + _options = options; + } + + // Straight to the registry: unlike Marten, Polecat's public EventStoreOptions facade has no + // AddEventType, so the shared IEventRegistry member on StoreOptions.EventGraph is the seam. + public void AddEventType(Type eventType) => _options.EventGraph.AddEventType(eventType); + + public ITagTypeRegistration RegisterTagType(string tableSuffix) where TTag : notnull + => _options.Events.RegisterTagType(tableSuffix); + + public void Snapshot(SnapshotLifecycle lifecycle) where TDoc : notnull + => _options.Projections.Snapshot(lifecycle); + + // Live aggregators are derived automatically -- see SupportsLiveAggregationRegistration. + public void LiveAggregation() where TDoc : notnull + { + } + } + + internal class PolecatComplianceBatch : IComplianceBatch + { + private readonly IBatchedQuery _batch; + + public PolecatComplianceBatch(IBatchedQuery batch) + { + _batch = batch; + } + + public Task EventsExist(EventTagQuery query) => _batch.EventsExist(query); + + public Task> FetchForWritingByTags(EventTagQuery query) where T : class + => _batch.FetchForWritingByTags(query); + + public Task Execute(CancellationToken token = default) => _batch.Execute(token); + } +} diff --git a/src/Polecat.Tests/Compliance/polecat_event_store_compliance.cs b/src/Polecat.Tests/Compliance/polecat_event_store_compliance.cs new file mode 100644 index 0000000..ea91c0d --- /dev/null +++ b/src/Polecat.Tests/Compliance/polecat_event_store_compliance.cs @@ -0,0 +1,23 @@ +using JasperFx.Events.ComplianceTests; + +namespace Polecat.Tests.Compliance; + +/* + * Polecat'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 Polecat's IEventStore session pair through + * PolecatComplianceFixture. Marten enrolls the same way, so drift between the two products' copies + * of these tests is no longer possible. + */ + +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/Polecat.Tests/Events/assign_tag_where_tests.cs b/src/Polecat.Tests/Events/assign_tag_where_tests.cs deleted file mode 100644 index 68064f7..0000000 --- a/src/Polecat.Tests/Events/assign_tag_where_tests.cs +++ /dev/null @@ -1,211 +0,0 @@ -#nullable enable -using JasperFx.Events; -using JasperFx.Events.Tags; -using Polecat.Tests.Harness; - -namespace Polecat.Tests.Events; - -public record RegionId(Guid Value); - -public record OrderPlaced(string OrderNumber, decimal Amount); -public record OrderShipped(string OrderNumber); -public record OrderCancelled(string OrderNumber, string Reason); - -public class assign_tag_where_tests : OneOffConfigurationsContext -{ - private RegionId _eastRegion = null!; - private RegionId _westRegion = null!; - - private async Task SetupStoreAsync() - { - _eastRegion = new RegionId(Guid.NewGuid()); - _westRegion = new RegionId(Guid.NewGuid()); - - ConfigureStore(opts => - { - opts.Events.RegisterTagType("region"); - }); - - await theDatabase.ApplyAllConfiguredChangesToDatabaseAsync(); - } - - [Fact] - public async Task assign_tag_where_by_event_type_name() - { - await SetupStoreAsync(); - - // Append events WITHOUT tags - await using var session1 = theStore.LightweightSession(); - var stream1 = Guid.NewGuid(); - session1.Events.Append(stream1, - new OrderPlaced("ORD-1", 100m), - new OrderShipped("ORD-1")); - await session1.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Now retroactively tag all OrderPlaced events with a region - await using var session2 = theStore.LightweightSession(); - var orderPlacedTypeName = theStore.Options.EventGraph.EventMappingFor(typeof(OrderPlaced)).EventTypeName; - session2.Events.AssignTagWhere( - e => e.EventTypeName == orderPlacedTypeName, - _eastRegion); - await session2.SaveChangesAsync(TestContext.Current.CancellationToken); - - // 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, TestContext.Current.CancellationToken); - - events.Count.ShouldBe(1); - events[0].Data.ShouldBeOfType().OrderNumber.ShouldBe("ORD-1"); - } - - [Fact] - public async Task assign_tag_where_by_stream_id() - { - await SetupStoreAsync(); - - var stream1 = Guid.NewGuid(); - var stream2 = Guid.NewGuid(); - - await using var session1 = theStore.LightweightSession(); - session1.Events.Append(stream1, - new OrderPlaced("ORD-1", 100m), - new OrderShipped("ORD-1")); - session1.Events.Append(stream2, - new OrderPlaced("ORD-2", 200m)); - await session1.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Tag all events in stream1 only - await using var session2 = theStore.LightweightSession(); - session2.Events.AssignTagWhere( - e => e.StreamId == stream1, - _eastRegion); - await session2.SaveChangesAsync(TestContext.Current.CancellationToken); - - // 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, TestContext.Current.CancellationToken); - - events.Count.ShouldBe(2); - events[0].Data.ShouldBeOfType().OrderNumber.ShouldBe("ORD-1"); - events[1].Data.ShouldBeOfType().OrderNumber.ShouldBe("ORD-1"); - } - - [Fact] - public async Task assign_tag_where_with_compound_predicate() - { - await SetupStoreAsync(); - - var stream1 = Guid.NewGuid(); - - await using var session1 = theStore.LightweightSession(); - session1.Events.Append(stream1, - new OrderPlaced("ORD-1", 100m), - new OrderShipped("ORD-1"), - new OrderCancelled("ORD-1", "changed mind")); - await session1.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Tag events that are of type OrderPlaced or OrderCancelled - await using var session2 = theStore.LightweightSession(); - var placedType = theStore.Options.EventGraph.EventMappingFor(typeof(OrderPlaced)).EventTypeName; - var cancelledType = theStore.Options.EventGraph.EventMappingFor(typeof(OrderCancelled)).EventTypeName; - - session2.Events.AssignTagWhere( - e => e.EventTypeName == placedType || e.EventTypeName == cancelledType, - _eastRegion); - await session2.SaveChangesAsync(TestContext.Current.CancellationToken); - - // 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, TestContext.Current.CancellationToken); - - 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() - { - await SetupStoreAsync(); - - var stream1 = Guid.NewGuid(); - - await using var session1 = theStore.LightweightSession(); - session1.Events.Append(stream1, new OrderPlaced("ORD-1", 100m)); - await session1.SaveChangesAsync(TestContext.Current.CancellationToken); - - var placedType = theStore.Options.EventGraph.EventMappingFor(typeof(OrderPlaced)).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(TestContext.Current.CancellationToken); - - await using var session3 = theStore.LightweightSession(); - session3.Events.AssignTagWhere( - e => e.EventTypeName == placedType, _eastRegion); - await session3.SaveChangesAsync(TestContext.Current.CancellationToken); - - // 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, TestContext.Current.CancellationToken); - events.Count.ShouldBe(1); - } - - [Fact] - public async Task assign_tag_where_does_not_affect_unmatched_events() - { - await SetupStoreAsync(); - - var stream1 = Guid.NewGuid(); - var stream2 = Guid.NewGuid(); - - await using var session1 = theStore.LightweightSession(); - session1.Events.Append(stream1, new OrderPlaced("ORD-1", 100m)); - session1.Events.Append(stream2, new OrderPlaced("ORD-2", 200m)); - await session1.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Only tag events in stream1 - await using var session2 = theStore.LightweightSession(); - session2.Events.AssignTagWhere( - e => e.StreamId == stream1, _eastRegion); - await session2.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Tag events in stream2 with different region - await using var session3 = theStore.LightweightSession(); - session3.Events.AssignTagWhere( - e => e.StreamId == stream2, _westRegion); - await session3.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Verify east only has stream1's event - await using var session4 = theStore.LightweightSession(); - var eastEvents = await session4.Events.QueryByTagsAsync( - new EventTagQuery().Or(_eastRegion), TestContext.Current.CancellationToken); - eastEvents.Count.ShouldBe(1); - eastEvents[0].Data.ShouldBeOfType().OrderNumber.ShouldBe("ORD-1"); - - // Verify west only has stream2's event - var westEvents = await session4.Events.QueryByTagsAsync( - new EventTagQuery().Or(_westRegion), TestContext.Current.CancellationToken); - westEvents.Count.ShouldBe(1); - westEvents[0].Data.ShouldBeOfType().OrderNumber.ShouldBe("ORD-2"); - } - - [Fact] - public async Task assign_tag_where_throws_for_unregistered_tag_type() - { - await SetupStoreAsync(); - - await using var session = theStore.LightweightSession(); - Should.Throw(() => - { - session.Events.AssignTagWhere(e => e.Sequence > 0, new StudentId(Guid.NewGuid())); - }); - } -} diff --git a/src/Polecat.Tests/Events/auto_discover_aggregate_types.cs b/src/Polecat.Tests/Events/auto_discover_aggregate_types.cs index 7d35717..4a01571 100644 --- a/src/Polecat.Tests/Events/auto_discover_aggregate_types.cs +++ b/src/Polecat.Tests/Events/auto_discover_aggregate_types.cs @@ -1,3 +1,4 @@ +using JasperFx.Events.ComplianceTests; using Polecat.Tests.Harness; namespace Polecat.Tests.Events; @@ -30,7 +31,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(TestContext.Current.CancellationToken); var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); diff --git a/src/Polecat.Tests/Events/dcb_documentation_samples.cs b/src/Polecat.Tests/Events/dcb_documentation_samples.cs new file mode 100644 index 0000000..24dbb30 --- /dev/null +++ b/src/Polecat.Tests/Events/dcb_documentation_samples.cs @@ -0,0 +1,302 @@ +#nullable enable +using JasperFx.Events; +using JasperFx.Events.Tags; +using Polecat.Events.Dcb; +using Polecat.Tests.Harness; + +namespace Polecat.Tests.Events; + +#region sample_polecat_dcb_tag_type_definitions +// Strong-typed tag identifiers +public record StudentId(Guid Value); +public record CourseId(Guid Value); +#endregion + +#region sample_polecat_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_polecat_dcb_aggregate +// Aggregate for DCB +public partial 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 Polecat's other DCB test fixtures share. +/// +/// +/// The behavioral coverage that used to live here now runs once in +/// +/// against every Critter Stack event store, so it can no longer drift from Marten's copy. What +/// stays behind is the documentation: each test below backs a sample_polecat_dcb_* snippet +/// block, so it has to keep compiling and passing with Polecat-flavored API calls in it. +/// +/// Polecat always uses Quick append (direct INSERT with OUTPUT seq_id). Tags are inserted +/// immediately after each event, so DCB works with the only append mode available. +/// +[Collection("integration")] +public class dcb_documentation_samples : IntegrationContext +{ + public dcb_documentation_samples(DefaultStoreFixture fixture) : base(fixture) + { + } + + #region sample_polecat_dcb_registering_tag_types + public override async ValueTask InitializeAsync() + { + await StoreOptions(opts => + { + // Register tag types -- each gets its own table (pc_event_tag_student, pc_event_tag_course) + opts.Events.RegisterTagType("student") + .ForAggregate(); + opts.Events.RegisterTagType("course") + .ForAggregate(); + }); + } + #endregion + + [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_polecat_dcb_tagging_events + var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); + enrolled.WithTag(studentId, courseId); + theSession.Events.Append(streamId, enrolled); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + #endregion + + await using var session2 = theStore.LightweightSession(); + #region sample_polecat_dcb_query_by_single_tag + var query = new EventTagQuery().Or(studentId); + var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); + #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(); + + var e1 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); + e1.WithTag(student1, course); + theSession.Events.Append(stream1, e1); + + var e2 = theSession.Events.BuildEvent(new StudentEnrolled("Bob", "Math")); + e2.WithTag(student2, course); + theSession.Events.Append(stream2, e2); + + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + await using var session2 = theStore.LightweightSession(); + #region sample_polecat_dcb_query_multiple_tags_or + var query = new EventTagQuery() + .Or(student1) + .Or(student2); + + var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); + #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(TestContext.Current.CancellationToken); + + await using var session2 = theStore.LightweightSession(); + #region sample_polecat_dcb_query_by_event_type + var query = new EventTagQuery() + .Or(studentId); + + var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); + #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(TestContext.Current.CancellationToken); + + await using var session2 = theStore.LightweightSession(); + #region sample_polecat_dcb_aggregate_by_tags + var query = new EventTagQuery() + .Or(studentId) + .Or(courseId); + + var aggregate = await session2.Events.AggregateByTagsAsync(query, TestContext.Current.CancellationToken); + #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(); + + var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); + enrolled.WithTag(studentId, courseId); + theSession.Events.Append(streamId, enrolled); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + #region sample_polecat_dcb_fetch_for_writing_by_tags + await using var session2 = theStore.LightweightSession(); + var query = new EventTagQuery().Or(studentId); + var boundary = await session2.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); + + // 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(TestContext.Current.CancellationToken); + #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(); + + var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); + enrolled.WithTag(studentId, courseId); + theSession.Events.Append(streamId, enrolled); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + // Session 1: fetch for writing + await using var session1 = theStore.LightweightSession(); + var query = new EventTagQuery().Or(studentId); + var boundary = await session1.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); + + // 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(TestContext.Current.CancellationToken); + + // 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_polecat_dcb_handling_concurrency + try + { + await session1.SaveChangesAsync(TestContext.Current.CancellationToken); + } + catch (AggregateException ex) when (ex.InnerExceptions.OfType().Any()) + { + // Reload and retry -- the boundary's tag query had new matching events + var violation = ex.InnerExceptions.OfType().First(); + // violation.Query -- the original tag query + // violation.LastSeenSequence -- the sequence at time of read + } + #endregion + } + + #region sample_polecat_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(TestContext.Current.CancellationToken); + + // Check existence -- lightweight, no event loading + await using var session2 = theStore.LightweightSession(); + var query = new EventTagQuery().Or(studentId); + var exists = await session2.Events.EventsExistAsync(query, TestContext.Current.CancellationToken); + exists.ShouldBeTrue(); + } + #endregion +} diff --git a/src/Polecat.Tests/Events/dcb_tag_query_and_consistency_tests.cs b/src/Polecat.Tests/Events/dcb_tag_query_and_consistency_tests.cs deleted file mode 100644 index 15f5944..0000000 --- a/src/Polecat.Tests/Events/dcb_tag_query_and_consistency_tests.cs +++ /dev/null @@ -1,771 +0,0 @@ -#nullable enable -using JasperFx.Events; -using JasperFx.Events.Tags; -using Polecat.Events.Dcb; -using Polecat.Tests.Harness; - -namespace Polecat.Tests.Events; - -#region sample_polecat_dcb_tag_type_definitions -// Strong-typed tag identifiers -public record StudentId(Guid Value); -public record CourseId(Guid Value); -#endregion - -#region sample_polecat_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_polecat_dcb_aggregate -// Aggregate for DCB -public partial 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 - -// Polecat always uses Quick append (direct INSERT with OUTPUT seq_id). -// Tags are inserted immediately after each event, so DCB works with the only append mode available. -[Collection("integration")] -public class dcb_tag_query_and_consistency_tests : IntegrationContext -{ - public dcb_tag_query_and_consistency_tests(DefaultStoreFixture fixture) : base(fixture) - { - } - - #region sample_polecat_dcb_registering_tag_types - public override async ValueTask InitializeAsync() - { - await StoreOptions(opts => - { - // Register tag types -- each gets its own table (pc_event_tag_student, pc_event_tag_course) - opts.Events.RegisterTagType("student") - .ForAggregate(); - opts.Events.RegisterTagType("course") - .ForAggregate(); - }); - } - #endregion - - [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_polecat_dcb_tagging_events - var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); - enrolled.WithTag(studentId, courseId); - theSession.Events.Append(streamId, enrolled); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - #endregion - - await using var session2 = theStore.LightweightSession(); - #region sample_polecat_dcb_query_by_single_tag - var query = new EventTagQuery().Or(studentId); - var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); - #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(); - - var e1 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); - e1.WithTag(student1, course); - theSession.Events.Append(stream1, e1); - - var e2 = theSession.Events.BuildEvent(new StudentEnrolled("Bob", "Math")); - e2.WithTag(student2, course); - theSession.Events.Append(stream2, e2); - - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - #region sample_polecat_dcb_query_multiple_tags_or - var query = new EventTagQuery() - .Or(student1) - .Or(student2); - - var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); - #endregion - events.Count.ShouldBe(2); - } - - [Fact] - public async Task can_query_events_across_distinct_tag_types_with_or() - { - // The core DCB boundary case: events on different streams carry DIFFERENT single tags, and the - // query OR-combines distinct tag types. Each matching event carries only one of the queried tag - // types — a regression guard against the INNER JOIN bug that required every event to carry all - // queried tag types (which collapsed this query to zero rows). - var studentId = new StudentId(Guid.NewGuid()); - var courseId = new CourseId(Guid.NewGuid()); - - // Tagged with ONLY the student - var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); - enrolled.WithTag(studentId); - theSession.Events.Append(Guid.NewGuid(), enrolled); - - // Tagged with ONLY the course - var submitted = theSession.Events.BuildEvent(new AssignmentSubmitted("HW1", 95)); - submitted.WithTag(courseId); - theSession.Events.Append(Guid.NewGuid(), submitted); - - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery() - .Or(studentId) - .Or(courseId); - - var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); - events.Count.ShouldBe(2); - events.ShouldContain(e => e.Data is StudentEnrolled); - events.ShouldContain(e => e.Data is AssignmentSubmitted); - } - - [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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - #region sample_polecat_dcb_query_by_event_type - var query = new EventTagQuery() - .Or(studentId); - - var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); - #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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(otherStudentId); - var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); - 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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - #region sample_polecat_dcb_aggregate_by_tags - var query = new EventTagQuery() - .Or(studentId) - .Or(courseId); - - var aggregate = await session2.Events.AggregateByTagsAsync(query, TestContext.Current.CancellationToken); - #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()); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var aggregate = await session2.Events.AggregateByTagsAsync(query, TestContext.Current.CancellationToken); - 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(); - - var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); - enrolled.WithTag(studentId, courseId); - theSession.Events.Append(streamId, enrolled); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - #region sample_polecat_dcb_fetch_for_writing_by_tags - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var boundary = await session2.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); - - // 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(TestContext.Current.CancellationToken); - #endregion - - boundary.Aggregate.ShouldNotBeNull(); - boundary.Aggregate!.StudentName.ShouldBe("Alice"); - boundary.Events.Count.ShouldBe(1); - } - - [Fact] - public async Task fetch_for_writing_appends_to_existing_tag_derived_stream_without_collision() - { - // Seed the student's stream using the tag value as the stream id, so the boundary's - // tag-derived routing targets this PRE-EXISTING stream on save. Before the fix the boundary - // used StreamAction.Start here (TryFindStream only sees the current session's pending work), - // throwing ExistingStreamIdCollisionException. - var studentId = new StudentId(Guid.NewGuid()); - - var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); - enrolled.WithTag(studentId); - theSession.Events.Append(studentId.Value, enrolled); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var boundary = await session2.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); - - var submitted = session2.Events.BuildEvent(new AssignmentSubmitted("HW1", 95)); - submitted.WithTag(studentId); - boundary.AppendOne(submitted); - - await Should.NotThrowAsync(async () => await session2.SaveChangesAsync()); - - await using var session3 = theStore.LightweightSession(); - var events = await session3.Events.QueryByTagsAsync(new EventTagQuery().Or(studentId), TestContext.Current.CancellationToken); - events.ShouldContain(e => e.Data is AssignmentSubmitted); - } - - [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(); - - var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); - enrolled.WithTag(studentId, courseId); - theSession.Events.Append(streamId, enrolled); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Session 1: fetch for writing - await using var session1 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var boundary = await session1.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); - - // 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(TestContext.Current.CancellationToken); - - // 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_polecat_dcb_handling_concurrency - try - { - await session1.SaveChangesAsync(TestContext.Current.CancellationToken); - } - catch (AggregateException ex) when (ex.InnerExceptions.OfType().Any()) - { - // Reload and retry -- the boundary's tag query had new matching events - var violation = ex.InnerExceptions.OfType().First(); - // violation.Query -- the original tag query - // violation.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(); - - var enrolled1 = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math")); - enrolled1.WithTag(student1, course); - theSession.Events.Append(stream1, enrolled1); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - // 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, TestContext.Current.CancellationToken); - - // 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(TestContext.Current.CancellationToken); - - // Session 1: save should succeed - var assignment = session1.Events.BuildEvent(new AssignmentSubmitted("HW1", 95)); - assignment.WithTag(student1, course); - boundary.AppendOne(assignment); - - await session1.SaveChangesAsync(TestContext.Current.CancellationToken); // 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(); - - 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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); - - 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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var events = await session2.Events.QueryByTagsAsync(query, TestContext.Current.CancellationToken); - - events.Count.ShouldBe(3); - 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, TestContext.Current.CancellationToken); - - 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(TestContext.Current.CancellationToken); - - // 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); - - var ex = await Should.ThrowAsync(async () => - { - await session1.SaveChangesAsync(); - }); - ex.InnerExceptions.ShouldContain(e => e is DcbConcurrencyException); - } - - [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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var batch = session2.CreateBatchQuery(); - var query = new EventTagQuery().Or(studentId); - var boundaryTask = batch.FetchForWritingByTags(query); - await batch.Execute(TestContext.Current.CancellationToken); - - 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(TestContext.Current.CancellationToken); - } - - [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(TestContext.Current.CancellationToken); - - // 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.FetchForWritingByTags(query); - await batch.Execute(TestContext.Current.CancellationToken); - 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(TestContext.Current.CancellationToken); - - // Session 1: try to save — should throw - var assignment = session1.Events.BuildEvent(new AssignmentSubmitted("HW1", 95)); - assignment.WithTag(studentId, courseId); - boundary.AppendOne(assignment); - - var ex = await Should.ThrowAsync(async () => - { - await session1.SaveChangesAsync(); - }); - ex.InnerExceptions.ShouldContain(e => e is DcbConcurrencyException); - } - - #region sample_polecat_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(TestContext.Current.CancellationToken); - - // Check existence -- lightweight, no event loading - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var exists = await session2.Events.EventsExistAsync(query, TestContext.Current.CancellationToken); - exists.ShouldBeTrue(); - } - #endregion - - [Fact] - public async Task events_exist_returns_false_when_no_matching_events() - { - var studentId = new StudentId(Guid.NewGuid()); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var exists = await session2.Events.EventsExistAsync(query, TestContext.Current.CancellationToken); - 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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - - // Should find StudentEnrolled - var query1 = new EventTagQuery().Or(studentId); - (await session2.Events.EventsExistAsync(query1, TestContext.Current.CancellationToken)).ShouldBeTrue(); - - // Should NOT find AssignmentSubmitted (none appended) - var query2 = new EventTagQuery().Or(studentId); - (await session2.Events.EventsExistAsync(query2, TestContext.Current.CancellationToken)).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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var batch = session2.CreateBatchQuery(); - var query = new EventTagQuery().Or(studentId); - var existsTask = batch.EventsExist(query); - await batch.Execute(TestContext.Current.CancellationToken); - - (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.EventsExist(query); - await batch.Execute(TestContext.Current.CancellationToken); - - (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(TestContext.Current.CancellationToken); - - // Fetch for writing - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var boundary = await session2.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); - - // 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(TestContext.Current.CancellationToken); - - // Verify the event is discoverable by tag query - await using var session3 = theStore.LightweightSession(); - var events = await session3.Events.QueryByTagsAsync( - new EventTagQuery().Or(studentId), TestContext.Current.CancellationToken); - 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(TestContext.Current.CancellationToken); - - // Fetch for writing - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var boundary = await session2.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); - - // 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(TestContext.Current.CancellationToken); - - // Fetch for writing - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var boundary = await session2.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); - - // 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(TestContext.Current.CancellationToken); - } - - [Fact] - public async Task append_event_with_tag_having_no_aggregate_type_creates_new_stream() - { - // Register a tag type WITHOUT an aggregate association - await StoreOptions(opts => - { - opts.Events.RegisterTagType("student"); - // CourseId registered WITHOUT ForAggregate - opts.Events.RegisterTagType("course"); - }); - - 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(TestContext.Current.CancellationToken); - - await using var session2 = theStore.LightweightSession(); - var query = new EventTagQuery().Or(studentId); - var boundary = await session2.Events.FetchForWritingByTags(query, TestContext.Current.CancellationToken); - - // 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(TestContext.Current.CancellationToken); - } -} diff --git a/src/Polecat.Tests/Events/self_aggregating_evolve_method.cs b/src/Polecat.Tests/Events/self_aggregating_evolve_method.cs deleted file mode 100644 index 0d25189..0000000 --- a/src/Polecat.Tests/Events/self_aggregating_evolve_method.cs +++ /dev/null @@ -1,316 +0,0 @@ -using JasperFx.Events; -using JasperFx.Events.Projections; -using Polecat.Projections; -using Polecat.Tests.Harness; - -namespace Polecat.Tests.Events; - -#region sample_polecat_evolve_aggregates - -public record AEvent; -public record BEvent; -public record CEvent; - -/// -/// 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 - -[Collection("integration")] -public class self_aggregating_evolve_method : IntegrationContext -{ - public self_aggregating_evolve_method(DefaultStoreFixture fixture) : base(fixture) - { - } - - [Fact] - public async Task mutable_ievent_evolve_inline() - { - await StoreOptions(opts => - { - opts.DatabaseSchemaName = "evolve_tests"; - opts.Projections.Add>(ProjectionLifecycle.Inline); - }); - - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new AEvent(), new CEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(2); - aggregate.BCount.ShouldBe(1); - aggregate.CCount.ShouldBe(1); - } - - [Fact] - public async Task mutable_object_evolve_inline() - { - await StoreOptions(opts => - { - opts.DatabaseSchemaName = "evolve_tests"; - opts.Projections.Add>(ProjectionLifecycle.Inline); - }); - - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new CEvent(), new CEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(1); - aggregate.BCount.ShouldBe(1); - aggregate.CCount.ShouldBe(2); - } - - [Fact] - public async Task immutable_ievent_evolve_inline() - { - await StoreOptions(opts => - { - opts.DatabaseSchemaName = "evolve_tests"; - opts.Projections.Add>(ProjectionLifecycle.Inline); - }); - - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new AEvent(), new AEvent(), new BEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(2); - aggregate.BCount.ShouldBe(1); - aggregate.CCount.ShouldBe(0); - } - - [Fact] - public async Task immutable_object_evolve_inline() - { - await StoreOptions(opts => - { - opts.DatabaseSchemaName = "evolve_tests"; - opts.Projections.Add>(ProjectionLifecycle.Inline); - }); - - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new BEvent(), new CEvent(), new AEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(1); - aggregate.BCount.ShouldBe(1); - aggregate.CCount.ShouldBe(1); - } - - [Fact] - public async Task async_evolve_inline() - { - await StoreOptions(opts => - { - opts.DatabaseSchemaName = "evolve_tests"; - opts.Projections.Add>(ProjectionLifecycle.Inline); - }); - - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new AEvent(), new AEvent(), new BEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(2); - aggregate.BCount.ShouldBe(1); - } - - [Fact] - public async Task immutable_async_evolve_inline() - { - await StoreOptions(opts => - { - opts.DatabaseSchemaName = "evolve_tests"; - opts.Projections.Add>(ProjectionLifecycle.Inline); - }); - - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new AEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(2); - aggregate.BCount.ShouldBe(1); - } - - [Fact] - public async Task mutable_ievent_evolve_with_append_to_existing_stream() - { - await StoreOptions(opts => - { - opts.DatabaseSchemaName = "evolve_tests"; - opts.Projections.Add>(ProjectionLifecycle.Inline); - }); - - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new AEvent(), new BEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - // Append more events - theSession.Events.Append(streamId, new AEvent(), new CEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(2); - aggregate.BCount.ShouldBe(1); - aggregate.CCount.ShouldBe(1); - } - - [Fact] - public async Task live_aggregation_with_evolve() - { - // No snapshot — live aggregation only - var streamId = Guid.NewGuid(); - theSession.Events.StartStream(streamId, new AEvent(), new BEvent(), new CEvent()); - await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); - - var aggregate = await theSession.Events.AggregateStreamAsync(streamId, token: TestContext.Current.CancellationToken); - aggregate.ShouldNotBeNull(); - aggregate.ACount.ShouldBe(1); - aggregate.BCount.ShouldBe(1); - aggregate.CCount.ShouldBe(1); - } -} diff --git a/src/Polecat.Tests/Polecat.Tests.csproj b/src/Polecat.Tests/Polecat.Tests.csproj index 3b3969e..1714f2f 100644 --- a/src/Polecat.Tests/Polecat.Tests.csproj +++ b/src/Polecat.Tests/Polecat.Tests.csproj @@ -28,6 +28,9 @@ + + diff --git a/src/Polecat.Tests/Projections/projection_sg_dispatch_audit_tests.cs b/src/Polecat.Tests/Projections/projection_sg_dispatch_audit_tests.cs index ad32ca8..7b67556 100644 --- a/src/Polecat.Tests/Projections/projection_sg_dispatch_audit_tests.cs +++ b/src/Polecat.Tests/Projections/projection_sg_dispatch_audit_tests.cs @@ -1,4 +1,5 @@ using JasperFx; +using JasperFx.Events.ComplianceTests; using JasperFx.Events.Projections; using Polecat.Projections; using Polecat.Tests.Events; @@ -59,7 +60,7 @@ public class projection_sg_dispatch_audit_tests Row("StringQuestAggregate (string)", opts => opts.Projections.Add>(ProjectionLifecycle.Inline)), Row("StudentCourseEnrollment (Guid)", - opts => opts.Projections.Add>(ProjectionLifecycle.Inline)), + opts => opts.Projections.Add>(ProjectionLifecycle.Inline)), Row("QuestAggregate (Guid)", opts => opts.Projections.Add>(ProjectionLifecycle.Inline)), Row("OrderAggregate (Guid, has natural key)", diff --git a/src/Polecat/Events/EventOperations.cs b/src/Polecat/Events/EventOperations.cs index 58d77ae..63a1448 100644 --- a/src/Polecat/Events/EventOperations.cs +++ b/src/Polecat/Events/EventOperations.cs @@ -941,7 +941,7 @@ public async Task> QueryByTagsAsync(EventTagQuery query, { var seqId = reader.GetInt64(0); var eventId = reader.GetGuid(1); - // stream_id at index 2 + var rawStreamId = reader.GetValue(2); var eventVersion = reader.GetInt64(3); var json = reader.GetString(4); var typeName = reader.GetString(5); @@ -966,6 +966,18 @@ public async Task> QueryByTagsAsync(EventTagQuery query, @event.DotNetTypeName = dotNetTypeName!; @event.IsArchived = isArchived; + // The stream_id column was selected but never mapped onto the envelope, so every event + // a DCB tag query returned carried StreamId == Guid.Empty. Hydrated the same way the + // daemon's event loader does. + if (_events.StreamIdentity == StreamIdentity.AsGuid && rawStreamId is Guid streamGuid) + { + @event.StreamId = streamGuid; + } + else + { + @event.StreamKey = rawStreamId.ToString(); + } + var metaIndex = 10; if (eventOptions.EnableCorrelationId) {