Fix on-the-fly event store schema + InitialData seeding on startup (#219)

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;
+
+/// <summary>
+/// #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.
+/// </summary>
+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<SeededDoc>().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<IHostedService>().ToList();
+        hostedServices.OfType<PolecatActivator>().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<IDocumentStore>();
+        await using var query = store.QuerySession();
+        var doc = await query.LoadAsync<SeededDoc>(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();
+    }
+}

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;
+
+/// <summary>
+/// #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".
+/// </summary>
+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<Exception>(session.SaveChangesAsync());
+    }
+}

src/Polecat/Events/QueryEventStore.cs

@@ -59,6 +59,7 @@ public async Task<IReadOnlyList<IEvent>> FetchStreamAsync(string streamKey, long
     private async Task<IReadOnlyList<IEvent>> 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<IReadOnlyList<IEvent>> FetchStreamInternalAsync(object stream
 
     private async Task<IEvent?> 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<IReadOnlyList<IEvent>> FetchStreamInternalAsync(object stream
 
     private async Task<StreamState?> 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

src/Polecat/Internal/DocumentSessionBase.cs

@@ -362,6 +362,14 @@
             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
@@ -992,9 +1000,9 @@
     {
         if (document is ITracked tracked)
         {
             tracked.CorrelationId = CorrelationId;
             tracked.CausationId = CausationId;
             tracked.LastModifiedBy = LastModifiedBy;
         }
 
         if (document is Metadata.ITenanted tenanted)

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<DocumentProvider> providers, Can
             await EnsureTableAsync(provider, token);
         }
     }
+
+    private volatile bool _eventStoreEnsured;
+    private readonly SemaphoreSlim _eventStoreSemaphore = new(1, 1);
+
+    /// <summary>
+    ///     #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.
+    /// </summary>
+    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<JasperFx.Events.Aggregation.IAggregateProjection>()
+                .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();
+        }
+    }
 }

src/Polecat/Internal/QuerySession.cs

@@ -153,6 +153,12 @@ public Task<bool> CheckExistsAsync<T>(string id, CancellationToken token = defau
     public Task<bool> CheckExistsAsync<T>(int id, CancellationToken token = default) where T : class
         => CheckExistsInternalAsync<T>(id, token);
 
+    /// <summary>
+    ///     #219: ensure the event store schema exists on the fly before an event read/write.
+    /// </summary>
+    internal Task EnsureEventStoreSchemaAsync(CancellationToken token)
+        => _tableEnsurer.EnsureEventStoreSchemaAsync(token);
+
     public Task<bool> CheckExistsAsync<T>(long id, CancellationToken token = default) where T : class
         => CheckExistsInternalAsync<T>(id, token);
 

src/Polecat/PolecatConfigurationExpression.cs

@@ -132,16 +132,25 @@ public PolecatConfigurationExpression ApplyAllDatabaseChangesOnStartup()
         return this;
     }
 
-    private void EnsureActivatorIsRegistered()
+    private void EnsureActivatorIsRegistered() => EnsureActivatorIsRegistered(Services);
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    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));
     }

src/Polecat/PolecatServiceCollectionExtensions.cs

@@ -109,6 +109,11 @@ public static PolecatConfigurationExpression AddPolecat(
         services.AddScoped(sp => sp.GetRequiredService<ISessionFactory>().OpenSession());
         services.AddScoped(sp => sp.GetRequiredService<ISessionFactory>().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);
     }
 }