From c6b1a93d2cbe3b222772bf75d1405bcd02ae4ad3 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Thu, 8 Oct 2026 08:58:53 -0500 Subject: [PATCH] GH-985: InMemoryDocumentStore implements IEventStore The in-memory prototyping store now implements IEventStore and is registered as IEventStore by AddInMemoryStoreForPrototyping, so the store-agnostic helpers written against IEventStore (Bobcat's GivenEvents/ThenEvents, services.EventStore()) work on a stub-first app. - AllDatabases() returns a single InMemoryEventDatabase; OpenSession opens a lightweight session on the store - OpenReadOnlyEventStore() is a query session's Events, which now also implements IReadOnlyEventStore: QueryEventsAsync (every filter but tags, which are refused) and QueryStreamStates - Projections are inline, so the database is never stale and reports no async shards; daemon, subscription and compaction members refuse with NotSupportedException - The compliance fixture's EventStore now returns the store Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J8KxC18Q51z5w4u6YARwe4 --- .../Documents/InMemoryIEventStoreTests.cs | 197 +++++++++++++++ .../InMemoryComplianceFixture.cs | 5 +- .../InMemoryDocumentStore.EventStore.cs | 228 ++++++++++++++++++ .../InMemoryEventDatabase.cs | 64 +++++ .../InMemoryEventOperations.cs | 16 +- .../InMemoryStoreRegistration.cs | 6 +- 6 files changed, 511 insertions(+), 5 deletions(-) create mode 100644 src/EventStoreTests/Documents/InMemoryIEventStoreTests.cs create mode 100644 src/JasperFx.Events.InMemory/InMemoryDocumentStore.EventStore.cs create mode 100644 src/JasperFx.Events.InMemory/InMemoryEventDatabase.cs diff --git a/src/EventStoreTests/Documents/InMemoryIEventStoreTests.cs b/src/EventStoreTests/Documents/InMemoryIEventStoreTests.cs new file mode 100644 index 00000000..20a6b55a --- /dev/null +++ b/src/EventStoreTests/Documents/InMemoryIEventStoreTests.cs @@ -0,0 +1,197 @@ +using EventStoreTests.Documents.LiveAggregation; +using JasperFx.Events; +using JasperFx.Events.Documents; +using JasperFx.Events.InMemory; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; + +namespace EventStoreTests.Documents; + +/// +/// jasperfx#985: the in-memory prototyping store is an , so the store-agnostic helpers +/// written against it -- Bobcat's arrange and assert steps first among them -- work on a stub-first app. +/// +/// +/// Each fact goes only through the contract, the same path Bobcat takes: a session +/// opened through IEventStore<,>.OpenSession on the first of AllDatabases(), appends through +/// , reads through . +/// +public class InMemoryIEventStoreTests +{ + private static CancellationToken Token => TestContext.Current.CancellationToken; + + private static async Task appendAsync(IEventStore store, Guid id, params object[] events) + { + var typed = (IEventStore)store; + var databases = await store.AllDatabases(); + + await using var session = typed.OpenSession(databases[0]); + IEventOperations operations = session.Events; + + var existing = await store.OpenReadOnlyEventStore().FetchStreamAsync(id, token: Token); + if (existing.Count == 0) operations.StartStream(id, events); + else operations.Append(id, events); + + await session.SaveChangesAsync(Token); + } + + [Fact] + public async Task the_registration_resolves_the_store_as_the_event_store() + { + using var host = Host.CreateDefaultBuilder() + .UseEnvironment(Environments.Development) + .ConfigureServices(services => services.AddInMemoryStoreForPrototyping()) + .Build(); + + var store = host.Services.GetRequiredService(); + store.ShouldBeSameAs(host.Services.GetRequiredService()); + + // One database, so a session can be opened on it + (await store.AllDatabases()).ShouldHaveSingleItem().ShouldBeOfType(); + store.DatabaseCardinality.ShouldBe(JasperFx.Descriptors.DatabaseCardinality.Single); + } + + [Fact] + public async Task arrange_then_read_back_through_the_contract() + { + IEventStore store = new InMemoryDocumentStore(); + var id = Guid.NewGuid(); + + await appendAsync(store, id, new TicketOpened("printer on fire")); + await appendAsync(store, id, new TicketAssigned("agent-7")); + + var events = await store.OpenReadOnlyEventStore().FetchStreamAsync(id, token: Token); + events.Select(x => x.Data.GetType()).ShouldBe([typeof(TicketOpened), typeof(TicketAssigned)]); + + // The read-only view is the session's IQueryEventStore, as on Marten, so it aggregates directly + var aggregated = await store.OpenReadOnlyEventStore().ShouldBeAssignableTo()! + .AggregateStreamAsync(id, token: Token); + aggregated!.Title.ShouldBe("printer on fire"); + aggregated.AssignedTo.ShouldBe("agent-7"); + } + + [Fact] + public async Task query_events_since_a_sequence_floor() + { + IEventStore store = new InMemoryDocumentStore(); + var one = Guid.NewGuid(); + var two = Guid.NewGuid(); + + await appendAsync(store, one, new TicketOpened("one")); + var database = (await store.AllDatabases())[0]; + var floor = await database.FetchHighestEventSequenceNumber(Token); + floor.ShouldBe(1); + + await appendAsync(store, two, new TicketOpened("two")); + await appendAsync(store, one, new TicketAssigned("agent-1")); + + // Bobcat asks for "everything since" with an unbounded page + var page = await store.OpenReadOnlyEventStore() + .QueryEventsAsync(new EventQuery { SequenceFloor = floor + 1, PageSize = int.MaxValue }, Token); + + page.TotalCount.ShouldBe(2); + page.Events.Select(x => x.Sequence).ShouldBe([2L, 3L]); + page.Events.Select(x => x.StreamId).ShouldBe([two, one]); + } + + [Fact] + public async Task query_events_filters_and_pages() + { + var store = new InMemoryDocumentStore(); + IEventStore events = store; + var one = Guid.NewGuid(); + var two = Guid.NewGuid(); + + await appendAsync(store, one, new TicketOpened("one")); + await appendAsync(store, two, new TicketOpened("two")); + await appendAsync(store, one, new TicketAssigned("agent-1")); + await appendAsync(store, two, new TicketAssigned("agent-2")); + await appendAsync(store, one, new TicketAssigned("agent-3")); + + var reader = events.OpenReadOnlyEventStore(); + + var byStream = await reader.QueryEventsAsync(new EventQuery { StreamId = one.ToString(), PageSize = 100 }, Token); + byStream.Events.ShouldAllBe(x => x.StreamId == one); + byStream.TotalCount.ShouldBe(3); + + var assigned = store.Events.EventMappingFor(typeof(TicketAssigned)).EventTypeName; + var byType = await reader.QueryEventsAsync( + new EventQuery { EventTypeName = assigned, StreamId = two.ToString(), PageSize = 100 }, Token); + byType.Events.ShouldHaveSingleItem().Data.ShouldBe(new TicketAssigned("agent-2")); + + var secondPage = await reader.QueryEventsAsync(new EventQuery { PageNumber = 2, PageSize = 2 }, Token); + secondPage.TotalCount.ShouldBe(5); + secondPage.Events.Select(x => x.Sequence).ShouldBe([3L, 4L]); + } + + [Fact] + public async Task tag_filters_are_refused_rather_than_ignored() + { + IEventStore store = new InMemoryDocumentStore(); + + await Should.ThrowAsync(() => store.OpenReadOnlyEventStore() + .QueryEventsAsync(new EventQuery { TagValues = { ["ticket"] = "1" } }, Token)); + } + + [Fact] + public async Task stream_states_are_queryable_with_the_shared_terminators() + { + var store = new InMemoryDocumentStore(); + var one = Guid.NewGuid(); + var two = Guid.NewGuid(); + + await appendAsync(store, one, new TicketOpened("one"), new TicketAssigned("agent-1")); + await appendAsync(store, two, new TicketOpened("two")); + + var states = await ((IEventStore)store).OpenReadOnlyEventStore().QueryStreamStates() + .Where(x => x.Version == 2) + .ToListAsync(Token); + + var state = states.ShouldHaveSingleItem(); + state.Id.ShouldBe(one); + state.AggregateType.ShouldBe(typeof(Ticket)); + } + + [Fact] + public async Task string_identified_streams_filter_on_the_key() + { + var store = new InMemoryDocumentStore(); + store.Events.StreamIdentity = StreamIdentity.AsString; + + await using (var session = ((IEventStore)store) + .OpenSession(store.Database)) + { + session.Events.StartStream("ESC-1", new TicketOpened("a")); + session.Events.StartStream("ESC-2", new TicketOpened("b")); + await session.SaveChangesAsync(Token); + } + + var page = await ((IEventStore)store).OpenReadOnlyEventStore() + .QueryEventsAsync(new EventQuery { StreamId = "ESC-2" }, Token); + + page.Events.ShouldHaveSingleItem().StreamKey.ShouldBe("ESC-2"); + } + + [Fact] + public async Task projections_are_never_stale_and_there_are_no_async_shards() + { + var store = new InMemoryDocumentStore(); + await appendAsync(store, Guid.NewGuid(), new TicketOpened("one")); + + var database = (await ((IEventStore)store).AllDatabases())[0]; + await database.WaitForNonStaleProjectionDataAsync(TimeSpan.FromMilliseconds(1)); + (await database.AllProjectionProgress(Token)).ShouldBeEmpty(); + + ((IEventStore)store).AllShards().ShouldBeEmpty(); + } + + [Fact] + public async Task the_daemon_is_refused_with_a_message_that_says_why() + { + IEventStore store = new InMemoryDocumentStore(); + + var ex = await Should.ThrowAsync(async () => await store.BuildProjectionDaemonAsync()); + ex.Message.ShouldContain("in-memory prototyping store"); + } +} diff --git a/src/InMemoryEventStoreTests/InMemoryComplianceFixture.cs b/src/InMemoryEventStoreTests/InMemoryComplianceFixture.cs index fa231e4d..b5ea8c84 100644 --- a/src/InMemoryEventStoreTests/InMemoryComplianceFixture.cs +++ b/src/InMemoryEventStoreTests/InMemoryComplianceFixture.cs @@ -106,9 +106,10 @@ public override Task CleanEventDataAsync() public override bool SupportsFlatTableProjections => false; - // ---- out of scope for a prototyping store ---- + // jasperfx#985: the store is an IEventStore + public override IEventStore EventStore => Store; - public override IEventStore EventStore => throw notSupported(nameof(EventStore)); + // ---- out of scope for a prototyping store ---- public override IComplianceBatch CreateBatch(IInMemoryQuerySession session) => throw notSupported("Batched queries"); diff --git a/src/JasperFx.Events.InMemory/InMemoryDocumentStore.EventStore.cs b/src/JasperFx.Events.InMemory/InMemoryDocumentStore.EventStore.cs new file mode 100644 index 00000000..60b09b68 --- /dev/null +++ b/src/JasperFx.Events.InMemory/InMemoryDocumentStore.EventStore.cs @@ -0,0 +1,228 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using JasperFx.Descriptors; +using JasperFx.Events.Daemon; +using JasperFx.Events.Descriptors; +using JasperFx.Events.Projections; +using Microsoft.Extensions.Logging; + +namespace JasperFx.Events.InMemory; + +/// +/// The prototyping store as a (jasperfx#985), so the store-agnostic helpers written +/// against it -- Bobcat's arrange and assert steps, services.EventStore() -- work on a stub-first app +/// before it has chosen Marten, Polecat or Fisher. +/// +/// +/// +/// One database, opened as a session on this store; the read-only tier is a session's Events, as on +/// Marten and Polecat. Projections run inline inside the commit, so there are no async shards, nothing is +/// ever stale, and the daemon-facing members refuse with a . +/// +/// +public partial class InMemoryDocumentStore : IEventStore +{ + private InMemoryEventDatabase? _database; + + internal InMemoryEventDatabase Database => _database ??= new InMemoryEventDatabase(this); + + // ---- IEventStore ---- + + /// + public Task TryCreateUsage(CancellationToken token) + => Task.FromResult(new EventStoreUsage(Subject, this) + { + Database = new DatabaseUsage { Cardinality = DatabaseCardinality.Single } + }); + + /// + public ValueTask BuildProjectionDaemonAsync(string? tenantIdOrDatabaseIdentifier = null, + ILogger? logger = null) + => throw InMemoryEventRegistry.NotSupported("The async daemon"); + + /// + public ValueTask BuildProjectionDaemonAsync(DatabaseId id) + => throw InMemoryEventRegistry.NotSupported("The async daemon"); + + /// + public Meter Meter { get; } = new("JasperFx.Events.InMemory"); + + /// + public ActivitySource ActivitySource { get; } = new("JasperFx.Events.InMemory"); + + /// + public string MetricsPrefix => "inmemory"; + + /// + public DatabaseCardinality DatabaseCardinality => DatabaseCardinality.Single; + + /// Events are single-tenanted on the prototyping store. + public bool HasMultipleTenants => false; + + /// + public EventStoreIdentity Identity { get; } = new("main", "inmemory"); + + /// + public ValueTask> AllDatabases() + => ValueTask.FromResult>([Database]); + + /// + /// The read-only event store: a query session's , which is also the + /// session's , so a caller can aggregate through it as on Marten. + /// + public IReadOnlyEventStore OpenReadOnlyEventStore() => QuerySession().Events; + + /// + public Task CompactStreamAsync(Guid streamId, CancellationToken token = default) + => throw InMemoryEventRegistry.NotSupported("Compacting a stream"); + + /// + public Task CompactStreamAsync(string streamKey, CancellationToken token = default) + => throw InMemoryEventRegistry.NotSupported("Compacting a stream"); + + // ---- IEventStore ---- + + IEventRegistry IEventStore.Registry => Events; + + /// + public Type IdentityTypeForProjectedType(Type aggregateType) + => InMemoryAggregateIdentity.ResolveIdType(aggregateType, Events.StreamIdentity); + + /// + public string DefaultDatabaseName => InMemoryEventDatabase.Name; + + /// + public ErrorHandlingOptions ContinuousErrors { get; } = new(); + + /// + public ErrorHandlingOptions RebuildErrors { get; } = new(); + + /// + public ErrorHandlingOptions ErrorHandlingOptions(ShardExecutionMode mode) + => mode == ShardExecutionMode.Rebuild ? RebuildErrors : ContinuousErrors; + + /// None: every projection on the prototyping store is inline. + public IReadOnlyList> AllShards() => []; + + /// + public TimeProvider TimeProvider => TimeProvider.System; + + /// There is no schema to create. + public AutoCreate AutoCreateSchemaObjects => AutoCreate.None; + + /// + public IInMemoryDocumentSession OpenSession(IEventDatabase database) => LightweightSession(); + + /// + public IInMemoryDocumentSession OpenSession(IEventDatabase database, string tenantId) + => LightweightSession(tenantId); + + /// + public Task RewindSubscriptionProgressAsync(IEventDatabase database, string subscriptionName, + CancellationToken token, long? sequenceFloor) + => throw InMemoryEventRegistry.NotSupported("Subscriptions"); + + /// + public Task RewindAgentProgressAsync(IEventDatabase database, string shardName, CancellationToken token, + long sequenceFloor) + => throw InMemoryEventRegistry.NotSupported("The async daemon"); + + /// + public Task TeardownExistingProjectionStateAsync(IEventDatabase database, string subscriptionName, + CancellationToken token) + => throw InMemoryEventRegistry.NotSupported("The async daemon"); + + /// + public Task DeleteProjectionProgressAsync(IEventDatabase database, string subscriptionName, + CancellationToken token) + => throw InMemoryEventRegistry.NotSupported("The async daemon"); + + /// + public ValueTask> StartProjectionBatchAsync( + EventRange range, IEventDatabase database, ShardExecutionMode mode, AsyncOptions projectionOptions, + CancellationToken token) + => throw InMemoryEventRegistry.NotSupported("The async daemon"); + + /// + public IEventLoader BuildEventLoader(IEventDatabase database, ILogger loggerFactory, EventFilterable filtering, + AsyncOptions shardOptions) + => throw InMemoryEventRegistry.NotSupported("The async daemon"); + + // ---- the reads behind the read-only tier ---- + + // Everything but the tag (DCB) filters, which the prototyping store does not record + internal const EventQueryFilters SupportedEventQueryFilters = EventQueryFilters.All & ~EventQueryFilters.Tags; + + internal PagedEvents QueryEvents(EventQuery query) + { + query.AssertFiltersAreSupported(SupportedEventQueryFilters); + + var typeNames = query.CombinedEventTypeNames(); + + List matching; + using (EnterReadGate()) + { + matching = _events + .Where(x => typeNames.Count == 0 || typeNames.Contains(x.EventTypeName)) + .Where(x => query.StreamId is null || streamIdMatches(x, query.StreamId)) + .Where(x => query.CorrelationId is null || x.CorrelationId == query.CorrelationId) + .Where(x => query.CausationId is null || x.CausationId == query.CausationId) + .Where(x => query.UserName is null || x.UserName == query.UserName) + .Where(x => query.TenantId is null || x.TenantId == query.TenantId) + .Where(x => query.TimestampFrom is null || x.Timestamp >= query.TimestampFrom.Value) + .Where(x => query.TimestampTo is null || x.Timestamp <= query.TimestampTo.Value) + .Where(x => query.SequenceFloor is null || x.Sequence >= query.SequenceFloor.Value) + .Where(x => query.SequenceCeiling is null || x.Sequence <= query.SequenceCeiling.Value) + .OrderBy(x => x.Sequence) + .ToList(); + } + + // In long arithmetic, so a caller asking for "everything" with PageSize = int.MaxValue cannot overflow + var pageNumber = Math.Max(query.PageNumber, 1); + var pageSize = Math.Max(query.PageSize, 1); + var skip = (long)(pageNumber - 1) * pageSize; + + return new PagedEvents + { + Events = matching.Skip((int)Math.Min(skip, int.MaxValue)).Take(pageSize).ToList(), + TotalCount = matching.Count, + PageNumber = pageNumber, + PageSize = pageSize + }; + } + + private bool streamIdMatches(IEvent @event, string streamId) + => Events.StreamIdentity == StreamIdentity.AsGuid + ? Guid.TryParse(streamId, out var id) && @event.StreamId == id + : @event.StreamKey == streamId; + + internal IReadOnlyList StreamStates(string? tenantId) + { + using (EnterReadGate()) + { + return _streams.Values + .Where(x => tenantId is null || x.TenantId == tenantId) + .Select(x => x.Id is Guid id + ? new StreamState(id, x.Version, x.AggregateType, x.LastTimestamp, x.Created) + : new StreamState((string)x.Id, x.Version, x.AggregateType, x.LastTimestamp, x.Created)) + .ToList(); + } + } + + internal long HighestEventSequence() + { + using (EnterReadGate()) + { + return _events.Count == 0 ? 0 : _events[^1].Sequence; + } + } + + internal long? EventSequenceFloorAt(DateTimeOffset timestamp) + { + using (EnterReadGate()) + { + var below = _events.Where(x => x.Timestamp < timestamp).ToList(); + return below.Count == 0 ? null : below.Max(x => x.Sequence); + } + } +} diff --git a/src/JasperFx.Events.InMemory/InMemoryEventDatabase.cs b/src/JasperFx.Events.InMemory/InMemoryEventDatabase.cs new file mode 100644 index 00000000..05faf113 --- /dev/null +++ b/src/JasperFx.Events.InMemory/InMemoryEventDatabase.cs @@ -0,0 +1,64 @@ +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Microsoft.Extensions.Logging.Abstractions; + +namespace JasperFx.Events.InMemory; + +/// +/// The one database of the in-memory prototyping store (jasperfx#985) -- what +/// returns, so a store-agnostic caller can open a session on it +/// and wait on its projections. +/// +/// +/// Projections run inline inside the commit, so there are no async shards to report progress for and the data +/// is never stale. +/// +public sealed class InMemoryEventDatabase : IEventDatabase +{ + internal const string Name = "main"; + + private readonly InMemoryDocumentStore _store; + + internal InMemoryEventDatabase(InMemoryDocumentStore store) + { + _store = store; + } + + /// + public string Identifier => Name; + + /// + public Uri DatabaseUri { get; } = new("inmemory://main"); + + /// + public string StorageIdentifier => "inmemory"; + + /// + public ShardStateTracker Tracker { get; } = new(NullLogger.Instance) { DatabaseIdentifier = Name }; + + /// Inline projections are written in the commit, so there is nothing to wait for. + public Task WaitForNonStaleProjectionDataAsync(TimeSpan timeout) => Task.CompletedTask; + + /// There is no storage to create. + public Task EnsureStorageExistsAsync(Type storageType, CancellationToken token) => Task.CompletedTask; + + /// Zero for any name: there are no async shards on the prototyping store. + public Task ProjectionProgressFor(ShardName name, CancellationToken token = default) + => Task.FromResult(0L); + + /// Empty: there are no async shards on the prototyping store. + public Task> AllProjectionProgress(CancellationToken token = default) + => Task.FromResult>([]); + + /// + public Task FetchHighestEventSequenceNumber(CancellationToken token) + => Task.FromResult(_store.HighestEventSequence()); + + /// + public Task FindEventStoreFloorAtTimeAsync(DateTimeOffset timestamp, CancellationToken token) + => Task.FromResult(_store.EventSequenceFloorAt(timestamp)); + + /// + public Task StoreDeadLetterEventAsync(object storage, DeadLetterEvent deadLetterEvent, CancellationToken token) + => throw InMemoryEventRegistry.NotSupported("Dead letter events"); +} diff --git a/src/JasperFx.Events.InMemory/InMemoryEventOperations.cs b/src/JasperFx.Events.InMemory/InMemoryEventOperations.cs index 6f391c55..fdedaa78 100644 --- a/src/JasperFx.Events.InMemory/InMemoryEventOperations.cs +++ b/src/JasperFx.Events.InMemory/InMemoryEventOperations.cs @@ -12,7 +12,7 @@ namespace JasperFx.Events.InMemory; /// Out of scope for a prototyping store, and refused with a that says /// so: archiving, compaction, rewriting events, and the tag (DCB) queries. /// -public partial class InMemoryEventOperations : IEventStoreOperations +public partial class InMemoryEventOperations : IEventStoreOperations, IReadOnlyEventStore { private readonly InMemoryDocumentSession _session; private readonly Dictionary _streams = new(); @@ -254,6 +254,20 @@ public Task> FetchStreamAsync(string streamKey, long versi public Task LoadAsync(Guid id, CancellationToken token = default) => Task.FromResult(Store.EventById(id)); + /// + /// The cross-stream query (jasperfx#985). Every filter but the tag (DCB) ones, which are refused rather than + /// ignored. + /// + public Task QueryEventsAsync(EventQuery query, CancellationToken token = default) + => Task.FromResult(Store.QueryEvents(query)); + + /// + /// Every stream, queried with LINQ-to-objects over a snapshot (jasperfx#985). Nothing is ever archived or + /// compacted on the prototyping store. + /// + public IQueryable QueryStreamStates(string? tenantId = null) + => InMemoryDocumentQueryable.Wrap(Store.StreamStates(tenantId).AsQueryable()); + // ---- Out of scope for the prototyping store ---- /// diff --git a/src/JasperFx.Events.InMemory/InMemoryStoreRegistration.cs b/src/JasperFx.Events.InMemory/InMemoryStoreRegistration.cs index f4237a46..59f6a325 100644 --- a/src/JasperFx.Events.InMemory/InMemoryStoreRegistration.cs +++ b/src/JasperFx.Events.InMemory/InMemoryStoreRegistration.cs @@ -92,8 +92,9 @@ public static class InMemoryStoreRegistration /// store holds survives a restart. /// /// - /// The store is registered as a singleton and as the - /// the rest of the Critter Stack resolves. + /// The store is registered as a singleton , and as the + /// and the rest of the Critter Stack + /// resolves (jasperfx#985). /// /// public static IServiceCollection AddInMemoryStoreForPrototyping(this IServiceCollection services, @@ -105,6 +106,7 @@ public static IServiceCollection AddInMemoryStoreForPrototyping(this IServiceCol services.AddSingleton(options); services.TryAddSingleton(_ => options.BuildStore()); services.TryAddSingleton(s => s.GetRequiredService()); + services.TryAddSingleton(s => s.GetRequiredService()); // A hosted service, so the guard runs as the host starts -- before anything has written to a // store that is about to vanish -- and a refusal stops the host rather than logging past it.