Merge remote-tracking branch 'origin/develop' into feature/pg-enum-refactor

Author: hhvrc

Common/OpenShockDb/EmailOutboxQueries.cs

@@ -0,0 +1,36 @@
+using Microsoft.EntityFrameworkCore;
+using OpenShock.Common.Models;
+
+namespace OpenShock.Common.OpenShockDb;
+
+/// <summary>
+/// Reusable queries over the <see cref="EmailOutboxMessage"/> table. Kept here (rather than inline in the
+/// Cron delivery job) so the raw SQL is a single source of truth that integration tests can execute
+/// directly - a table or column rename then breaks the test rather than surfacing only in production.
+/// </summary>
+public static class EmailOutboxQueries
+{
+    /// <summary>How many due rows the delivery job claims per batch.</summary>
+    public const int ClaimBatchSize = 50;
+
+    /// <summary>
+    /// The delivery job's claim query: due rows - <see cref="EmailStatus.Pending"/> past their
+    /// next-attempt time, or <see cref="EmailStatus.Sending"/> whose lease has lapsed - oldest first,
+    /// capped at <paramref name="batchSize"/>, locked with <c>FOR UPDATE SKIP LOCKED</c> so concurrent
+    /// runs take disjoint batches. Run it inside an explicit transaction to hold the locks for the claim.
+    /// </summary>
+    public static IQueryable<EmailOutboxMessage> DueForDelivery(this DbSet<EmailOutboxMessage> outbox, int batchSize)
+    {
+        if (batchSize <= 0) throw new ArgumentOutOfRangeException(nameof(batchSize));
+
+        return outbox.FromSql(
+            $"""
+             SELECT * FROM email_outbox
+             WHERE next_attempt_at <= now()
+               AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending})
+             ORDER BY next_attempt_at
+             LIMIT {batchSize}
+             FOR UPDATE SKIP LOCKED
+             """);
+    }
+}

Common/OpenShockMiddlewareHelper.cs

@@ -118,4 +118,59 @@ public static async Task<IApplicationBuilder> ApplyPendingOpenShockMigrations(th
 
         return app;
     }
+
+    /// <summary>
+    /// Blocks startup until the database schema is up to date (no pending migrations), for hosts that do
+    /// <em>not</em> own migrations (e.g. the Cron worker - the API is the sole migrator). This is not just
+    /// tidiness: <see cref="OpenShockContext"/> binds Postgres enum types (<c>email_status</c>,
+    /// <c>email_type</c>, ...) to CLR enums <em>by name</em>, and Npgsql resolves those names against the
+    /// type catalog it loads on the data source's first connection and caches for the life of the process.
+    /// If that first connection happens before the migrator has created a newly-added enum, every query
+    /// using it fails permanently ("data type name '...' could not be found") until the process restarts.
+    /// Waiting here guarantees the schema - and its enum types - exists before anything opens the pooled
+    /// context, closing the deploy-time race without this host performing any schema writes of its own.
+    /// </summary>
+    public static async Task<IApplicationBuilder> WaitForOpenShockSchemaReady(this IApplicationBuilder app,
+        DatabaseOptions options, TimeSpan? timeout = null)
+    {
+        using var scope = app.ApplicationServices.CreateScope();
+        var loggerFactory = scope.ServiceProvider.GetRequiredService<ILoggerFactory>();
+        var logger = loggerFactory.CreateLogger("SchemaReadyGate");
+
+        logger.LogInformation("Waiting for database schema to be up to date (migrations owned by another host)...");
+
+        var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromMinutes(5));
+        var delay = TimeSpan.FromSeconds(1);
+        var maxDelay = TimeSpan.FromSeconds(15);
+
+        while (true)
+        {
+            try
+            {
+                await using var migrationContext = new MigrationOpenShockContext(options.Conn, options.Debug, loggerFactory);
+                var pending = (await migrationContext.Database.GetPendingMigrationsAsync()).ToArray();
+                if (pending.Length == 0)
+                {
+                    logger.LogInformation("Database schema is up to date; proceeding with startup");
+                    return app;
+                }
+
+                logger.LogWarning("Schema not ready: {Count} pending migration(s) [{Migrations}] not yet applied by the migrator",
+                    pending.Length, string.Join(", ", pending));
+            }
+            catch (Exception ex)
+            {
+                // Database not reachable yet, or the migrator is mid-run. Keep waiting rather than crash-looping.
+                logger.LogWarning(ex, "Could not determine migration state; will retry");
+            }
+
+            if (DateTime.UtcNow >= deadline)
+                throw new TimeoutException(
+                    "Timed out waiting for the database schema to be brought up to date by the migrator. " +
+                    "Ensure the API host (the sole migrator) is running and reachable.");
+
+            await Task.Delay(delay);
+            delay = TimeSpan.FromSeconds(Math.Min(maxDelay.TotalSeconds, delay.TotalSeconds * 2));
+        }
+    }
 }

Cron.IntegrationTests/Tests/EmailOutboxQueryTests.cs

@@ -0,0 +1,47 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using OpenShock.Common.Models;
+using OpenShock.Common.OpenShockDb;
+
+namespace OpenShock.Cron.IntegrationTests.Tests;
+
+/// <summary>
+/// Runs the delivery job's actual claim query (<see cref="EmailOutboxQueries.DueForDelivery"/>, the single
+/// source of truth the job uses) against real Postgres via the Cron host's own pooled context - the same
+/// registration + enum-mapping path that runs in production. Guards the raw SQL (table/column names, the
+/// email_status enum, LIMIT, FOR UPDATE SKIP LOCKED, SELECT * materialization) so a schema rename breaks
+/// this test rather than only surfacing in the Cron host at runtime.
+/// </summary>
+public sealed class EmailOutboxQueryTests
+{
+    [ClassDataSource<CronApplicationFactory>(Shared = SharedType.PerTestSession)]
+    public required CronApplicationFactory Factory { get; init; }
+
+    [Test]
+    public async Task DueForDelivery_RunsTheJobsRawSql_AgainstTheRealSchema()
+    {
+        var recipient = $"outbox-claim-sql-{Guid.CreateVersion7():N}@test.org";
+
+        Guid id;
+        await using (var db = await Factory.DbContextFactory.CreateDbContextAsync())
+        {
+            var message = EmailOutboxMessage.Create(
+                EmailType.PasswordReset, recipient, "Claim",
+                new Dictionary<string, string> { ["k"] = "v" });
+            // Dated well into the past so it sorts first and the global LIMIT can never exclude it.
+            message.NextAttemptAt = DateTime.UtcNow - TimeSpan.FromDays(1);
+            db.EmailOutbox.Add(message);
+            await db.SaveChangesAsync();
+            id = message.Id;
+        }
+
+        // Resolve the pooled OpenShockContext exactly as the Cron delivery job does (DI-injected) so this
+        // covers the same registration + enum-mapping path that runs in production.
+        using var scope = Factory.Services.CreateScope();
+        var db2 = scope.ServiceProvider.GetRequiredService<OpenShockContext>();
+
+        var claimed = await db2.EmailOutbox.DueForDelivery(1).ToListAsync();
+
+        await Assert.That(claimed.Any(m => m.Id == id)).IsTrue();
+    }
+}

Cron/Jobs/EmailOutboxDeliveryJob.cs

@@ -25,7 +25,7 @@ namespace OpenShock.Cron.Jobs;
 [CronJob("* * * * *")] // Every minute (https://crontab.guru/)
 public sealed class EmailOutboxDeliveryJob
 {
-    private const int BatchSize = 50;
+    private const int BatchSize = EmailOutboxQueries.ClaimBatchSize;
 
     private readonly OpenShockContext _db;
     private readonly IEmailOutboxDispatcher _dispatcher;
@@ -88,15 +88,7 @@ private async Task<List<Guid>> ClaimDueBatchAsync()
 
         await using var transaction = await _db.Database.BeginTransactionAsync();
 
-        var due = await _db.EmailOutbox.FromSql(
-            $"""
-             SELECT * FROM email_outbox
-             WHERE next_attempt_at <= now()
-               AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending})
-             ORDER BY next_attempt_at
-             LIMIT {BatchSize}
-             FOR UPDATE SKIP LOCKED
-             """).ToListAsync();
+        var due = await _db.EmailOutbox.DueForDelivery(BatchSize).ToListAsync();
 
         if (due.Count == 0)
         {

Cron/Program.cs

@@ -44,6 +44,13 @@
 
 await app.UseCommonOpenShockMiddleware();
 
+// The Cron host does not own migrations (the API is the sole migrator). Its OpenShockContext binds
+// Postgres enum types by name at the pooled data source's first connection and caches them for the
+// process's life, so it must not open that context before a newly-added enum exists - otherwise every
+// claim query fails permanently ("data type name 'email_status' could not be found"). Block until the
+// migrator has applied all pending migrations, which happens before Hangfire or any job runs below.
+await app.WaitForOpenShockSchemaReady(databaseOptions);
+
 var hangfireOptions = new DashboardOptions();
 if (app.Environment.IsProduction() || Environment.GetEnvironmentVariable("DOTNET_RUNNING_IN_CONTAINER") == "true")
 {