diff --git a/Common/OpenShockDb/EmailOutboxQueries.cs b/Common/OpenShockDb/EmailOutboxQueries.cs new file mode 100644 index 00000000..ade6d04a --- /dev/null +++ b/Common/OpenShockDb/EmailOutboxQueries.cs @@ -0,0 +1,36 @@ +using Microsoft.EntityFrameworkCore; +using OpenShock.Common.Models; + +namespace OpenShock.Common.OpenShockDb; + +/// +/// Reusable queries over the 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. +/// +public static class EmailOutboxQueries +{ + /// How many due rows the delivery job claims per batch. + public const int ClaimBatchSize = 50; + + /// + /// The delivery job's claim query: due rows - past their + /// next-attempt time, or whose lease has lapsed - oldest first, + /// capped at , locked with FOR UPDATE SKIP LOCKED so concurrent + /// runs take disjoint batches. Run it inside an explicit transaction to hold the locks for the claim. + /// + public static IQueryable DueForDelivery(this DbSet 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 + """); + } +} diff --git a/Cron.IntegrationTests/Tests/EmailOutboxQueryTests.cs b/Cron.IntegrationTests/Tests/EmailOutboxQueryTests.cs new file mode 100644 index 00000000..d69df67d --- /dev/null +++ b/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; + +/// +/// Runs the delivery job's actual claim query (, 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. +/// +public sealed class EmailOutboxQueryTests +{ + [ClassDataSource(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 { ["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(); + + var claimed = await db2.EmailOutbox.DueForDelivery(1).ToListAsync(); + + await Assert.That(claimed.Any(m => m.Id == id)).IsTrue(); + } +} diff --git a/Cron/Jobs/EmailOutboxDeliveryJob.cs b/Cron/Jobs/EmailOutboxDeliveryJob.cs index 59f45bef..c970be61 100644 --- a/Cron/Jobs/EmailOutboxDeliveryJob.cs +++ b/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> 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) {