diff --git a/src/Polecat.Tests/Seeding/initial_data_host_startup_tests.cs b/src/Polecat.Tests/Seeding/initial_data_host_startup_tests.cs new file mode 100644 index 00000000..c2d32ba6 --- /dev/null +++ b/src/Polecat.Tests/Seeding/initial_data_host_startup_tests.cs @@ -0,0 +1,95 @@ +using JasperFx; +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Polecat; +using Polecat.Internal; +using Polecat.Linq; +using Polecat.TestUtils; +using Shouldly; + +namespace Polecat.Tests.Seeding; + +/// +/// #219: IInitialData seeders, added via AddPolecat(opts => opts.InitialData.Add(...)), must run on +/// host startup WITHOUT having to call ApplyAllDatabaseChangesOnStartup. Previously the activator that +/// runs the seeders was only registered by ApplyAllDatabaseChangesOnStartup / AddAsyncDaemon / +/// AddProjectionCoordinator, so the reporter's seeding never executed. +/// +public class initial_data_host_startup_tests +{ + private const string Schema = "initial_data_host"; + private static readonly Guid SeededId = Guid.Parse("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); + + public class SeededDoc + { + public Guid Id { get; set; } + public string Name { get; set; } = string.Empty; + } + + // Mirrors marcominerva's seeder: query first (must create the table on the fly), then seed. + private class Seeder : IInitialData + { + public async Task Populate(IDocumentStore store, CancellationToken cancellation) + { + await using var session = store.LightweightSession(); + if (await session.Query().AnyAsync(cancellation)) return; + + session.Store(new SeededDoc { Id = SeededId, Name = "seeded" }); + await session.SaveChangesAsync(cancellation); + } + } + + [Fact] + public async Task initial_data_runs_on_host_startup_without_ApplyAllDatabaseChangesOnStartup() + { + await DropSchemaAsync(); + + var services = new ServiceCollection(); + services.AddPolecat(opts => + { + opts.ConnectionString = ConnectionSource.ConnectionString; + opts.DatabaseSchemaName = Schema; + opts.UseNativeJsonType = ConnectionSource.SupportsNativeJson; + opts.InitialData.Add(new Seeder()); + // NOTE: deliberately NOT calling ApplyAllDatabaseChangesOnStartup() + }); + + await using var provider = services.BuildServiceProvider(); + + // The activator that runs InitialData must be registered just by calling AddPolecat. + var hostedServices = provider.GetServices().ToList(); + hostedServices.OfType().ShouldHaveSingleItem(); + + // Simulate host startup. + foreach (var hosted in hostedServices) + { + await hosted.StartAsync(CancellationToken.None); + } + + // Seeding ran — and the document table was created on the fly by the seeder. + var store = provider.GetRequiredService(); + await using var query = store.QuerySession(); + var doc = await query.LoadAsync(SeededId); + doc.ShouldNotBeNull(); + doc!.Name.ShouldBe("seeded"); + } + + private static async Task DropSchemaAsync() + { + await using var conn = new SqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $""" + IF SCHEMA_ID('{Schema}') IS NOT NULL + BEGIN + DECLARE @sql NVARCHAR(MAX) = ''; + SELECT @sql += 'DROP TABLE IF EXISTS ' + QUOTENAME(s.name) + '.' + QUOTENAME(t.name) + ';' + CHAR(13) + FROM sys.tables t JOIN sys.schemas s ON t.schema_id = s.schema_id + WHERE s.name = '{Schema}'; + EXEC sp_executesql @sql; + END + """; + await cmd.ExecuteNonQueryAsync(); + } +} diff --git a/src/Polecat.Tests/Storage/on_the_fly_event_store_tests.cs b/src/Polecat.Tests/Storage/on_the_fly_event_store_tests.cs new file mode 100644 index 00000000..a0d16b05 --- /dev/null +++ b/src/Polecat.Tests/Storage/on_the_fly_event_store_tests.cs @@ -0,0 +1,56 @@ +using JasperFx; +using Polecat.Tests.Harness; + +namespace Polecat.Tests.Storage; + +/// +/// #219: like document tables, the event store schema (pc_streams / pc_events / pc_event_progression) +/// should be created on the fly on first usage — unless on-the-fly migration is disabled with +/// AutoCreate.None. Previously only ApplyAllDatabaseChangesOnStartup created the event tables, so a +/// first append/query on a fresh database failed with "invalid object name". +/// +public class on_the_fly_event_store_tests : OneOffConfigurationsContext +{ + public record ThingHappened(string Name); + + [Fact] + public async Task appending_events_on_a_fresh_database_creates_the_event_store_schema() + { + ConfigureStore(_ => { }); // no ApplyAllDatabaseChangesOnStartup + + var streamId = Guid.NewGuid(); + await using (var session = theStore.LightweightSession()) + { + session.Events.StartStream(streamId, new ThingHappened("first")); + await session.SaveChangesAsync(); + } + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId); + events.Count.ShouldBe(1); + } + + [Fact] + public async Task querying_events_on_a_fresh_database_creates_the_event_store_schema() + { + ConfigureStore(_ => { }); + + // No append yet — a query on a fresh DB must still not blow up on missing tables. + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(Guid.NewGuid()); + events.Count.ShouldBe(0); + } + + [Fact] + public async Task auto_create_none_does_not_create_the_event_store_on_the_fly() + { + // When the user opts out with AutoCreate.None, on-the-fly creation must NOT happen. + ConfigureStore(opts => opts.AutoCreateSchemaObjects = AutoCreate.None); + + await using var session = theStore.LightweightSession(); + session.Events.StartStream(Guid.NewGuid(), new ThingHappened("x")); + + // The append should fail because the tables were never created and we did not auto-create. + await Should.ThrowAsync(session.SaveChangesAsync()); + } +} diff --git a/src/Polecat/Events/QueryEventStore.cs b/src/Polecat/Events/QueryEventStore.cs index 66b0f80f..c7a2f08b 100644 --- a/src/Polecat/Events/QueryEventStore.cs +++ b/src/Polecat/Events/QueryEventStore.cs @@ -59,6 +59,7 @@ public async Task> FetchStreamAsync(string streamKey, long private async Task> FetchStreamInternalAsync(object streamId, long version, DateTimeOffset? timestamp, long fromVersion, CancellationToken token) { + await _session.EnsureEventStoreSchemaAsync(token); // #219: create event store on first use // #57 pc_events half: column projection + per-row hydration live in // PcEventsRowReader, shared with the IEventStore explorer's // ReadStreamAsync path. This method only composes WHERE / ORDER BY. @@ -142,6 +143,7 @@ private async Task> FetchStreamInternalAsync(object stream private async Task LoadInternalAsync(Guid id, CancellationToken token) { + await _session.EnsureEventStoreSchemaAsync(token); // #219: create event store on first use // Mirrors FetchStreamInternalAsync but filters by the event UUID // rather than stream id, and reads the row's stream_id column to // assemble the context (since the caller doesn't know the stream @@ -191,6 +193,7 @@ private async Task> FetchStreamInternalAsync(object stream private async Task FetchStreamStateInternalAsync(object streamId, CancellationToken token) { + await _session.EnsureEventStoreSchemaAsync(token); // #219: create event store on first use // #57: column projection + row read live in PcStreamsRowReader so this // method, GetRecentStreamsAsync, and GetStreamMetadataAsync all read // pc_streams with the same shape. Note the canonical column order diff --git a/src/Polecat/Internal/DocumentSessionBase.cs b/src/Polecat/Internal/DocumentSessionBase.cs index 544f3428..4cb99408 100644 --- a/src/Polecat/Internal/DocumentSessionBase.cs +++ b/src/Polecat/Internal/DocumentSessionBase.cs @@ -362,6 +362,14 @@ private async Task SaveChangesInternalAsync(CancellationToken token) await _tableEnsurer.EnsureTablesAsync(typesNeeded, token); } + // #219: ensure the event store schema exists before appending — the event-sourcing analogue + // of ensuring document tables above. Runs outside the data transaction (it opens its own + // connection) and only when there are events to write. + if (_workTracker.Streams.Any(s => s.Events.Any())) + { + await _tableEnsurer.EnsureEventStoreSchemaAsync(token); + } + await _transactional.BeginTransactionAsync(token); using var tx = _transactional.Transaction!; try diff --git a/src/Polecat/Internal/DocumentTableEnsurer.cs b/src/Polecat/Internal/DocumentTableEnsurer.cs index 079ccf13..f02abbbe 100644 --- a/src/Polecat/Internal/DocumentTableEnsurer.cs +++ b/src/Polecat/Internal/DocumentTableEnsurer.cs @@ -43,6 +43,14 @@ public async Task EnsureTableAsync(DocumentProvider provider, CancellationToken return; } + // #219: honor the user's explicit opt-out. AutoCreate.None means "I manage the schema", + // so never create/alter tables implicitly on first use (mirrors Marten). + if (_options.AutoCreateSchemaObjects == AutoCreate.None) + { + _ensured.TryAdd(docType, true); + return; + } + await _semaphore.WaitAsync(token); try { @@ -187,4 +195,53 @@ public async Task EnsureTablesAsync(IEnumerable providers, Can await EnsureTableAsync(provider, token); } } + + private volatile bool _eventStoreEnsured; + private readonly SemaphoreSlim _eventStoreSemaphore = new(1, 1); + + /// + /// #219: ensures the event store schema (streams / events / progression tables, plus any tag + /// and natural-key tables) exists on first usage of the event store — the event-sourcing + /// analogue of EnsureTableAsync for documents. Idempotent and applied once per process; a + /// no-op under AutoCreate.None so the user's manual-schema opt-out is respected. + /// + public async Task EnsureEventStoreSchemaAsync(CancellationToken token) + { + if (_eventStoreEnsured) return; + + if (_options.AutoCreateSchemaObjects == AutoCreate.None) + { + _eventStoreEnsured = true; + return; + } + + await _eventStoreSemaphore.WaitAsync(token); + try + { + if (_eventStoreEnsured) return; + + await using var conn = _connectionFactory.Create(); + await conn.OpenAsync(token); + + var migrator = new SqlServerMigrator(); + + // Mirror PolecatDatabase.BuildFeatureSchemas: the event store feature owns the natural-key + // tables for aggregate projections that declare one. + var naturalKeys = _options.Projections.All + .OfType() + .Where(p => p.NaturalKeyDefinition != null) + .Select(p => p.NaturalKeyDefinition!) + .ToList(); + + var eventSchema = new Events.Schema.EventStoreFeatureSchema(_options.EventGraph, naturalKeys); + var migration = await SchemaMigration.DetermineAsync(conn, token, eventSchema.Objects); + await migrator.ApplyAllAsync(conn, migration, AutoCreate.CreateOrUpdate, ct: token); + + _eventStoreEnsured = true; + } + finally + { + _eventStoreSemaphore.Release(); + } + } } diff --git a/src/Polecat/Internal/QuerySession.cs b/src/Polecat/Internal/QuerySession.cs index 0827bae7..59bfea5e 100644 --- a/src/Polecat/Internal/QuerySession.cs +++ b/src/Polecat/Internal/QuerySession.cs @@ -153,6 +153,12 @@ public Task CheckExistsAsync(string id, CancellationToken token = defau public Task CheckExistsAsync(int id, CancellationToken token = default) where T : class => CheckExistsInternalAsync(id, token); + /// + /// #219: ensure the event store schema exists on the fly before an event read/write. + /// + internal Task EnsureEventStoreSchemaAsync(CancellationToken token) + => _tableEnsurer.EnsureEventStoreSchemaAsync(token); + public Task CheckExistsAsync(long id, CancellationToken token = default) where T : class => CheckExistsInternalAsync(id, token); diff --git a/src/Polecat/PolecatConfigurationExpression.cs b/src/Polecat/PolecatConfigurationExpression.cs index f65380d7..c2eb8a24 100644 --- a/src/Polecat/PolecatConfigurationExpression.cs +++ b/src/Polecat/PolecatConfigurationExpression.cs @@ -132,16 +132,25 @@ public PolecatConfigurationExpression ApplyAllDatabaseChangesOnStartup() return this; } - private void EnsureActivatorIsRegistered() + private void EnsureActivatorIsRegistered() => EnsureActivatorIsRegistered(Services); + + /// + /// Registers the PolecatActivator hosted service exactly once. The activator applies schema + /// changes on startup (when ApplyAllDatabaseChangesOnStartup is set) and ALWAYS runs the + /// configured InitialData seeders — so #219: it is registered unconditionally by AddPolecat + /// so that InitialData runs even when the app never calls ApplyAllDatabaseChangesOnStartup / + /// AddAsyncDaemon / AddProjectionCoordinator. + /// + internal static void EnsureActivatorIsRegistered(IServiceCollection services) { - if (Services.Any(x => + if (services.Any(x => x.ServiceType == typeof(IHostedService) && x.ImplementationType == typeof(PolecatActivator))) { return; } - Services.Insert(0, + services.Insert(0, new ServiceDescriptor(typeof(IHostedService), typeof(PolecatActivator), ServiceLifetime.Singleton)); } diff --git a/src/Polecat/PolecatServiceCollectionExtensions.cs b/src/Polecat/PolecatServiceCollectionExtensions.cs index fbadbedd..1e5a75d4 100644 --- a/src/Polecat/PolecatServiceCollectionExtensions.cs +++ b/src/Polecat/PolecatServiceCollectionExtensions.cs @@ -109,6 +109,11 @@ public static PolecatConfigurationExpression AddPolecat( services.AddScoped(sp => sp.GetRequiredService().OpenSession()); services.AddScoped(sp => sp.GetRequiredService().QuerySession()); + // #219: register the activator unconditionally so InitialData seeders run on host startup + // even without ApplyAllDatabaseChangesOnStartup. StartAsync is a no-op when there is no + // InitialData and ShouldApplyChangesOnStartup is false, so this is safe for every app. + PolecatConfigurationExpression.EnsureActivatorIsRegistered(services); + return new PolecatConfigurationExpression(services); } }