KaliCZ · KaliCZ · Jul 29, 2026 · Jul 29, 2026
@@ -785,7 +785,7 @@ public async Task PostComment_NotifiesTheBlogAuthor()
         var response = await client.PostAsJsonAsync($"/api/blog/{slug}/comments", new { content }, Ct);
         Assert.Equal(HttpStatusCode.OK, response.StatusCode);
 
-        var email = Assert.Single(await WaitForEmailsAsync(m => m.TextBody.Value.Contains(content), expectedCount: 1));
+        var email = Assert.Single(await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(content)));
         Assert.Equal("author@kalandra.local", email.To.Address);
         Assert.Contains(slug, email.Subject.Value);
         Assert.Contains($"https://www.kalandra.tech/blog/{slug}", email.TextBody.Value);
@@ -804,7 +804,7 @@ public async Task Reply_NotifiesBlogAuthorAndParentCommentAuthor()
         Authenticate(userId: Guid.NewGuid(), email: "replier@test.com");
         await client.PostAsJsonAsync($"/api/blog/{slug}/comments", new { content = replyContent, parentCommentId = parentId }, Ct);
 
-        var emails = await WaitForEmailsAsync(m => m.TextBody.Value.Contains(replyContent), expectedCount: 2);
+        var emails = await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(replyContent));
         Assert.Equal(2, emails.Length);
         Assert.Contains(emails, m => m.To.Address == "author@kalandra.local");
         var parentNotification = Assert.Single(emails, m => m.To.Address == "parent-author@test.com");
@@ -825,10 +825,7 @@ await client.PostAsJsonAsync(
             new { content = replyContent, parentCommentId = parent.GetProperty("id").GetString() },
             Ct);
 
-        var emails = await WaitForEmailsAsync(m => m.TextBody.Value.Contains(replyContent), expectedCount: 1);
-        // Grace period: a wrong extra notification would arrive moments later.
-        await Task.Delay(1500, Ct);
-        emails = [.. factory.EmailSender.Sent.Where(m => m.TextBody.Value.Contains(replyContent))];
+        var emails = await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(replyContent));
 
         var email = Assert.Single(emails);
         Assert.Equal("author@kalandra.local", email.To.Address);
@@ -849,26 +846,11 @@ public async Task AuthorsOwnComment_SendsNoEmailAtAll()
         var comments = (await ParseJsonAsync(await client.GetAsync($"/api/blog/{slug}/comments", Ct))).GetProperty("comments");
         Assert.Equal(1, comments.GetArrayLength());
 
-        await Task.Delay(1500, Ct);
-        Assert.DoesNotContain(factory.EmailSender.Sent, m => m.TextBody.Value.Contains(content));
+        Assert.Empty(await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(content)));
     }
 
     // ───── Helpers ─────
 
-    private async Task<Kalandra.Infrastructure.Email.EmailMessage[]> WaitForEmailsAsync(
-        Func<Kalandra.Infrastructure.Email.EmailMessage, bool> predicate, int expectedCount)
-    {
-        var deadline = DateTime.UtcNow.AddSeconds(20);
-        while (DateTime.UtcNow < deadline)
-        {
-            var matches = factory.EmailSender.Sent.Where(predicate).ToArray();
-            if (matches.Length >= expectedCount)
-                return matches;
-            await Task.Delay(200, Ct);
-        }
-        return [.. factory.EmailSender.Sent.Where(predicate)];
-    }
-
     /// <summary>Unique per test — the factory shares one database across the class.</summary>
     private static string NewSlug() => $"post-{Guid.NewGuid():N}";
 

@@ -697,7 +697,7 @@ public async Task Create_NotifiesTheOwner()
         Assert.Equal(HttpStatusCode.Created, response.StatusCode);
         var offerId = AssertValidGuid(await ParseJsonAsync(response), "id");
 
-        var email = Assert.Single(await WaitForEmailsAsync(m => m.TextBody.Value.Contains(marker), expectedCount: 1));
+        var email = Assert.Single(await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(marker)));
         Assert.Equal("owner@kalandra.local", email.To.Address);
         Assert.Equal("New job offer: Senior Developer at Acme Corp", email.Subject.Value);
         Assert.Contains("john@acme.com", email.TextBody.Value);
@@ -713,10 +713,7 @@ public async Task OfferAuthorsComment_NotifiesOnlyTheOwner()
         var response = await client.PostAsJsonAsync($"/api/job-offers/{id}/comments", new { content = marker }, Ct);
         Assert.Equal(HttpStatusCode.OK, response.StatusCode);
 
-        var emails = await WaitForEmailsAsync(m => m.TextBody.Value.Contains(marker), expectedCount: 1);
-        // Grace period: a wrong extra notification would arrive moments later.
-        await Task.Delay(1500, Ct);
-        emails = [.. factory.EmailSender.Sent.Where(m => m.TextBody.Value.Contains(marker))];
+        var emails = await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(marker));
 
         var email = Assert.Single(emails);
         Assert.Equal("owner@kalandra.local", email.To.Address);
@@ -734,10 +731,7 @@ public async Task OwnersComment_NotifiesOnlyTheOfferAuthor()
         var response = await client.PostAsJsonAsync($"/api/job-offers/{id}/comments", new { content = marker }, Ct);
         Assert.Equal(HttpStatusCode.OK, response.StatusCode);
 
-        var emails = await WaitForEmailsAsync(m => m.TextBody.Value.Contains(marker), expectedCount: 1);
-        // Grace period: a wrong extra notification would arrive moments later.
-        await Task.Delay(1500, Ct);
-        emails = [.. factory.EmailSender.Sent.Where(m => m.TextBody.Value.Contains(marker))];
+        var emails = await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(marker));
 
         var email = Assert.Single(emails);
         Assert.Equal("notified-author@test.com", email.To.Address);
@@ -755,7 +749,7 @@ public async Task NonOwnerAdminsComment_NotifiesOwnerAndOfferAuthor()
         var response = await client.PostAsJsonAsync($"/api/job-offers/{id}/comments", new { content = marker }, Ct);
         Assert.Equal(HttpStatusCode.OK, response.StatusCode);
 
-        var emails = await WaitForEmailsAsync(m => m.TextBody.Value.Contains(marker), expectedCount: 2);
+        var emails = await factory.WaitForDeliveredEmailsAsync(m => m.TextBody.Value.Contains(marker));
         Assert.Equal(2, emails.Length);
         Assert.Contains(emails, m => m.To.Address == "owner@kalandra.local");
         Assert.Contains(emails, m => m.To.Address == "watched-author@test.com");
@@ -810,20 +804,6 @@ public async Task Health_ReturnsOk()
 
     private static readonly Guid AdminUserId = new("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa");
 
-    private async Task<Kalandra.Infrastructure.Email.EmailMessage[]> WaitForEmailsAsync(
-        Func<Kalandra.Infrastructure.Email.EmailMessage, bool> predicate, int expectedCount)
-    {
-        var deadline = DateTime.UtcNow.AddSeconds(20);
-        while (DateTime.UtcNow < deadline)
-        {
-            var matches = factory.EmailSender.Sent.Where(predicate).ToArray();
-            if (matches.Length >= expectedCount)
-                return matches;
-            await Task.Delay(200, Ct);
-        }
-        return [.. factory.EmailSender.Sent.Where(predicate)];
-    }
-
     private Guid Authenticate(
         string email = "test@example.com",
         bool isAdmin = false)

@@ -5,6 +5,9 @@
 using Kalandra.Infrastructure.Storage;
 using Kalandra.Infrastructure.Turnstile;
 using Kalandra.Infrastructure.Users;
+using Marten;
+using Marten.Events;
+using Marten.Events.Daemon.Coordination;
 using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Mvc.Testing;
@@ -38,10 +41,26 @@ public class TestWebApplicationFactory : WebApplicationFactory<Program>, IAsyncL
     private readonly PostgreSqlContainer _postgres = new PostgreSqlBuilder("postgres:17-alpine")
         .Build();
 
+    // Generous because CI runs Docker and several test hosts in parallel; progress-based waits return
+    // the moment the daemon catches up, so the cap only bites when something is genuinely broken.
+    private static readonly TimeSpan AsyncDaemonTimeout = TimeSpan.FromSeconds(60);
+
     public FakeSupabaseAdminService FakeAdminService { get; } = new();
     public TestEmailSender EmailSender { get; } = new();
     public FakeUserInfoService UserInfoService { get; } = new();
 
+    /// <summary>
+    /// Waits until the async daemon has processed every event committed so far, then returns the sent
+    /// emails matching the predicate. The result is final for those events — a missing or extra
+    /// notification is a real bug, not a race — so callers assert exact counts without sleeping.
+    /// </summary>
+    public async Task<EmailMessage[]> WaitForDeliveredEmailsAsync(Func<EmailMessage, bool> predicate)
+    {
+        var store = Services.GetRequiredService<IDocumentStore>();
+        await store.WaitForNonStaleProjectionDataAsync(AsyncDaemonTimeout);
+        return [.. EmailSender.Sent.Where(predicate)];
+    }
+
     protected override void ConfigureWebHost(IWebHostBuilder builder)
     {
         // Mirror the dev/e2e environment so the configs' prod-only localhost checks stay out of the test host.
@@ -106,6 +125,29 @@ protected override void ConfigureClient(HttpClient client)
     public async ValueTask InitializeAsync()
     {
         await _postgres.StartAsync();
+        await WaitForNotificationSubscriptionsToStartAsync();
+    }
+
+    // SubscribeFromPresent seeds each subscription at the current high-water mark when its shard starts —
+    // an event committed before that would be skipped, never emailed. Agents register only after seeding,
+    // so waiting for registration while the database is still empty guarantees no test event is missed.
+    private async Task WaitForNotificationSubscriptionsToStartAsync()
+    {
+        var store = (DocumentStore)Services.GetRequiredService<IDocumentStore>();
+        var expectedShards = store.Options.Projections.AllShards().Select(shard => shard.Name.Identity).ToHashSet();
+        var daemon = Services.GetRequiredService<IProjectionCoordinator>().DaemonForMainDatabase();
+
+        using var cancellation = new CancellationTokenSource(AsyncDaemonTimeout);
+        try
+        {
+            while (!expectedShards.IsSubsetOf(daemon.CurrentAgents().Select(agent => agent.Name.Identity)))
+                await Task.Delay(100, cancellation.Token);
+        }
+        catch (OperationCanceledException)
+        {
+            throw new TimeoutException(
+                $"The async daemon did not start subscriptions [{string.Join(", ", expectedShards)}] within {AsyncDaemonTimeout}.");
+        }
     }
 
     public new async ValueTask DisposeAsync()