test(infra): cover scheduler/hosted-service loop orchestration + idempotency (#243)

tests/Orbit.Infrastructure.Tests/Services/OpenAiBatchPollerServiceTests.cs

@@ -164,6 +164,54 @@ public async Task PollPendingBatches_CompletedBatch_FileDeletionFailure_IsSwallo
         harness.Logger.Entries.Should().Contain(e => e.Level == LogLevel.Warning && e.Message.Contains("batch_1"));
     }
 
+    [Fact]
+    public async Task PollPendingBatches_CompletedAndInFlightBatches_FinalizesCompletedAndLeavesInFlight()
+    {
+        var completedBatch = AiFactExtractionBatch.Create(UserId, "batch_done", "file_in_done");
+        var inFlightBatch = AiFactExtractionBatch.Create(UserId, "batch_pending", "file_in_pending");
+
+        var harness = new PollerHarness()
+            .WithPendingBatches(completedBatch, inFlightBatch)
+            .WithBatchStatus("batch_done", new BatchStatusResult("completed", "file_out_done", null))
+            .WithBatchOutput("file_out_done", BatchOutputJsonl("User works night shifts"))
+            .WithBatchStatus("batch_pending", new BatchStatusResult("in_progress", null, null));
+
+        await harness.Service.PollPendingBatches(CancellationToken.None);
+
+        completedBatch.Status.Should().Be(AiFactExtractionBatchStatus.Completed);
+        inFlightBatch.Status.Should().Be(AiFactExtractionBatchStatus.Submitted);
+        harness.AddedFacts.Should().ContainSingle().Which.FactText.Should().Be("User works night shifts");
+        await harness.BatchClient.Received(1).DownloadFileAsync("file_out_done", Arg.Any<CancellationToken>());
+        await harness.BatchClient.Received(1).DeleteFileAsync("file_in_done", Arg.Any<CancellationToken>());
+        await harness.BatchClient.DidNotReceive().DeleteFileAsync("file_in_pending", Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_HostedLifecycle_RunsOnePollTickThenStopsGracefully()
+    {
+        var batch = AiFactExtractionBatch.Create(UserId, "batch_1", "file_in_1");
+        var harness = new PollerHarness()
+            .WithPendingBatches(batch)
+            .WithBatchStatus("batch_1", new BatchStatusResult("completed", "file_out_1", null))
+            .WithBatchOutput("file_out_1", BatchOutputJsonl("User works night shifts"));
+
+        var tickSaved = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        harness.UnitOfWork.SaveChangesAsync(Arg.Any<CancellationToken>())
+            .Returns(_ => { tickSaved.TrySetResult(); return Task.FromResult(1); });
+
+        var service = harness.Service;
+
+        await service.StartAsync(CancellationToken.None);
+        await tickSaved.Task.WaitAsync(TimeSpan.FromSeconds(30));
+        await service.StopAsync(CancellationToken.None);
+
+        batch.Status.Should().Be(AiFactExtractionBatchStatus.Completed);
+        harness.AddedFacts.Should().ContainSingle();
+        service.ExecuteTask.Should().NotBeNull();
+        service.ExecuteTask!.IsCompleted.Should().BeTrue();
+        service.ExecuteTask!.IsFaulted.Should().BeFalse();
+    }
+
     private sealed class PollerHarness
     {
         public IAiBatchClient BatchClient { get; } = Substitute.For<IAiBatchClient>();

tests/Orbit.Infrastructure.Tests/Services/PlayNotificationCleanupServiceTests.cs

@@ -2,6 +2,7 @@
 using Microsoft.Data.Sqlite;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using Orbit.Domain.Entities;
 using Orbit.Infrastructure.Persistence;
@@ -78,6 +79,30 @@ public async Task RunAsync_NothingExpired_DeletesNothing()
         (await _dbContext.ProcessedStripeEvents.CountAsync()).Should().Be(1);
     }
 
+    [Fact]
+    public async Task ExecuteAsync_HostedLifecycle_RunsOnePurgePassThenStopsGracefully()
+    {
+        AddPlayNotification("play-old", DaysAgo(31));
+        AddPlayNotification("play-recent", DaysAgo(1));
+        AddStripeEvent("stripe-old", DaysAgo(91));
+        AddStripeEvent("stripe-recent", DaysAgo(1));
+        await _dbContext.SaveChangesAsync();
+
+        var logger = new SignalOnStripePurgeLogger();
+        var service = new PlayNotificationCleanupService(_scopeFactory, logger);
+
+        await service.StartAsync(CancellationToken.None);
+        await logger.StripePurged.Task.WaitAsync(TimeSpan.FromSeconds(30));
+        await service.StopAsync(CancellationToken.None);
+
+        (await _dbContext.ProcessedPlayNotifications.Select(n => n.MessageId).ToListAsync())
+            .Should().ContainSingle().Which.Should().Be("play-recent");
+        (await _dbContext.ProcessedStripeEvents.Select(e => e.EventId).ToListAsync())
+            .Should().ContainSingle().Which.Should().Be("stripe-recent");
+        service.ExecuteTask.Should().NotBeNull();
+        service.ExecuteTask!.IsFaulted.Should().BeFalse();
+    }
+
     private static DateTime DaysAgo(int days) => DateTime.UtcNow.AddDays(-days);
 
     private void AddPlayNotification(string messageId, DateTime processedAtUtc)
@@ -99,6 +124,36 @@ private static void SetProcessedAt(ProcessedExternalEvent target, DateTime proce
             .GetProperty(nameof(ProcessedExternalEvent.ProcessedAtUtc))!
             .SetValue(target, processedAtUtc);
 
+    /// <summary>
+    /// Completes <see cref="StripePurged"/> when the Stripe-events purge log (EventId 5) fires, which is
+    /// the tick's final DB operation. Waiting on that signal lets the test read the shared context only
+    /// after the purge finishes, avoiding a concurrent access with the background loop.
+    /// </summary>
+    private sealed class SignalOnStripePurgeLogger : ILogger<PlayNotificationCleanupService>
+    {
+        public TaskCompletionSource StripePurged { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(
+            LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+        {
+            if (eventId.Id == 5)
+                StripePurged.TrySetResult();
+        }
+
+        private sealed class NullScope : IDisposable
+        {
+            public static readonly NullScope Instance = new();
+
+            public void Dispose()
+            {
+            }
+        }
+    }
+
     private sealed class SqliteCompatOrbitDbContext(DbContextOptions<OrbitDbContext> options)
         : OrbitDbContext(options)
     {

tests/Orbit.Infrastructure.Tests/Services/StreakFreezeAutoActivationServiceTests.cs

@@ -428,6 +428,79 @@ await pushService.Received(1).SendToUserAsync(
         }
     }
 
+    [Fact]
+    public async Task ActivateMissedDayFreezes_ExistingFreezeForMissedDate_DoesNotDoubleActivate()
+    {
+        var user = CreateEligibleProUser();
+        var missedDate = DateOnly.FromDateTime(DateTime.UtcNow).AddDays(-1);
+
+        await using var dbContext = CreateInMemoryDbContext();
+        var pushService = Substitute.For<IPushNotificationService>();
+
+        dbContext.Users.Add(user);
+        dbContext.StreakFreezes.Add(StreakFreeze.Create(user.Id, missedDate));
+        await dbContext.SaveChangesAsync();
+
+        var service = CreateService(dbContext, pushService);
+        await service.ActivateMissedDayFreezes(CancellationToken.None);
+
+        (await dbContext.StreakFreezes.AsNoTracking().CountAsync(f => f.UserId == user.Id)).Should().Be(1);
+        user.StreakFreezesAccumulated.Should().Be(1);
+        await pushService.DidNotReceive().SendToUserAsync(
+            user.Id, Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task ActivateMissedDayFreezes_GuardAlreadyRecordedForMissedDate_DoesNotReactivateOrRepush()
+    {
+        var user = CreateEligibleProUser();
+        var missedDate = DateOnly.FromDateTime(DateTime.UtcNow).AddDays(-1);
+
+        await using var dbContext = CreateInMemoryDbContext();
+        var pushService = Substitute.For<IPushNotificationService>();
+
+        dbContext.Users.Add(user);
+        dbContext.SentStreakFreezeAlerts.Add(SentStreakFreezeAlert.Create(user.Id, missedDate));
+        await dbContext.SaveChangesAsync();
+
+        var service = CreateService(dbContext, pushService);
+        await service.ActivateMissedDayFreezes(CancellationToken.None);
+
+        (await dbContext.StreakFreezes.AsNoTracking().CountAsync(f => f.UserId == user.Id)).Should().Be(0);
+        user.StreakFreezesAccumulated.Should().Be(1);
+        await pushService.DidNotReceive().SendToUserAsync(
+            user.Id, Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_HostedLifecycle_RunsOneFreezePassThenStopsGracefully()
+    {
+        var user = CreateEligibleProUser();
+
+        await using var dbContext = CreateInMemoryDbContext();
+        var pushService = Substitute.For<IPushNotificationService>();
+
+        var firstPush = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        pushService
+            .SendToUserAsync(user.Id, Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+            .Returns(_ => { firstPush.TrySetResult(); return Task.CompletedTask; });
+
+        dbContext.Users.Add(user);
+        await dbContext.SaveChangesAsync();
+
+        var service = CreateService(dbContext, pushService);
+
+        await service.StartAsync(CancellationToken.None);
+        await firstPush.Task.WaitAsync(TimeSpan.FromSeconds(30));
+        await service.StopAsync(CancellationToken.None);
+
+        (await dbContext.StreakFreezes.AsNoTracking().CountAsync(f => f.UserId == user.Id)).Should().Be(1);
+        user.StreakFreezesAccumulated.Should().Be(0);
+        service.ExecuteTask.Should().NotBeNull();
+        service.ExecuteTask!.IsCompleted.Should().BeTrue();
+        service.ExecuteTask!.IsFaulted.Should().BeFalse();
+    }
+
     private static User CreateEligibleProUser()
     {
         var user = User.Create($"User-{Guid.NewGuid():N}", $"{Guid.NewGuid():N}@test.com").Value;

tests/Orbit.Infrastructure.Tests/Services/XpAwardLogBackfillHostedServiceTests.cs

@@ -117,6 +117,40 @@ public async Task RunBackfillAsync_ScopeCreationThrows_LogsErrorAndDoesNotRethro
         _logger.Entries.Should().ContainSingle(e => e.Level == LogLevel.Error && e.EventId == 3);
     }
 
+    [Fact]
+    public async Task RunBackfillAsync_RunTwiceOnRestart_SecondRunSkipsAndDoesNotDoubleBackfill()
+    {
+        await SeedUserAsync(50);
+        var sut = CreateSut();
+
+        await sut.RunBackfillAsync(CancellationToken.None);
+        await sut.RunBackfillAsync(CancellationToken.None);
+
+        await using var verify = CreateContext(_dbName);
+        (await verify.XpAwardLogs.CountAsync()).Should().Be(1);
+        (await verify.AppConfigs.CountAsync(c => c.Key == BackfillFlag)).Should().Be(1);
+        _logger.Entries.Should().Contain(e => e.Level == LogLevel.Information && e.EventId == 2);
+        _logger.Entries.Should().Contain(e => e.Level == LogLevel.Information && e.EventId == 1);
+    }
+
+    [Fact]
+    public async Task StartAsync_CancelledDuringStartupDelay_StopsCleanlyWithoutRunningBackfill()
+    {
+        await SeedUserAsync(50);
+        var sut = CreateSut();
+
+        await sut.StartAsync(CancellationToken.None);
+        await sut.StopAsync(CancellationToken.None);
+
+        await using var verify = CreateContext(_dbName);
+        (await verify.XpAwardLogs.CountAsync()).Should().Be(0);
+        (await verify.AppConfigs.AnyAsync(c => c.Key == BackfillFlag)).Should().BeFalse();
+        _logger.Entries.Should().BeEmpty();
+        sut.ExecuteTask.Should().NotBeNull();
+        sut.ExecuteTask!.IsCompleted.Should().BeTrue();
+        sut.ExecuteTask!.IsFaulted.Should().BeFalse();
+    }
+
     private XpAwardLogBackfillHostedService CreateSut(IServiceScopeFactory? scopeFactory = null) =>
         new(scopeFactory ?? _scopeFactory, _logger);