feat(kpi): GetLatestRollupHourAsync watermark seam for the backfill fast-path (plan R2-04 T2)

src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs

@@ -109,6 +109,16 @@ public interface IKpiHistoryRepository
         string source, string metric, string scope, string? scopeKey,
         DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Returns the newest <c>KpiRollupHourly.HourStartUtc</c> across all series, or
+    /// <c>null</c> when no rollups exist. The recorder's one-shot backfill consults
+    /// this watermark to skip (or shrink to) the un-rolled tail instead of re-folding
+    /// the whole raw-retention window on every failover (arch-review 04 round 2, R1).
+    /// </summary>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>A task resolving to the newest folded hour start, or <c>null</c> when empty.</returns>
+    Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Bulk-deletes hourly rollup rows whose
     /// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi.KpiRollupHourly.HourStartUtc"/> is

src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs

@@ -192,6 +192,15 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
             .ToListAsync(cancellationToken);
     }
 
+    /// <inheritdoc />
+    public async Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default)
+    {
+        // Single MAX over the unique IX_KpiRollupHourly_Series population (the rollup
+        // table is small — one row per series-hour). Null when no rollups exist.
+        return await _context.KpiRollupHourly
+            .MaxAsync(r => (DateTime?)r.HourStartUtc, cancellationToken);
+    }
+
     /// <inheritdoc />
     public async Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default)
     {

tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs

@@ -406,6 +406,29 @@ public class KpiHistoryRepositoryTests
         Assert.Empty(ctx.ChangeTracker.Entries<KpiSample>());
     }
 
+    [Fact]
+    public async Task GetLatestRollupHour_ReturnsNull_WhenNoRollupsExist()
+    {
+        await using var ctx = NewContext();
+        var repo = new KpiHistoryRepository(ctx);
+        Assert.Null(await repo.GetLatestRollupHourAsync());
+    }
+
+    [Fact]
+    public async Task GetLatestRollupHour_ReturnsNewestHourStart_AcrossAllSeries()
+    {
+        await using var ctx = NewContext();
+        var repo = new KpiHistoryRepository(ctx);
+        await repo.RecordSamplesAsync(new[]
+        {
+            Sample("NotificationOutbox", "queueDepth", "Global", null, 5, Base.AddMinutes(10)),
+            Sample("SiteCallAudit", "buffered", "Global", null, 2, Base.AddHours(3).AddMinutes(10)),
+        });
+        await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(4));
+
+        Assert.Equal(Base.AddHours(3), await repo.GetLatestRollupHourAsync());
+    }
+
     // Local mirror of the repo's private hour-truncation, for cutoff assertions.
     private static DateTime TruncateHour(DateTime utc) =>
         new(utc.Year, utc.Month, utc.Day, utc.Hour, 0, 0, DateTimeKind.Utc);

tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs

@@ -191,6 +191,9 @@ public class KpiHistoryRecorderActorTests : TestKit
             lock (_gate) { _rollupPurgeCutoff = before; }
             return Task.CompletedTask;
         }
+
+        public Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
+            Task.FromResult<DateTime?>(null);
     }
 
     /// <summary>
@@ -237,6 +240,9 @@ public class KpiHistoryRecorderActorTests : TestKit
 
         public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
             Task.CompletedTask;
+
+        public Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
+            Task.FromResult<DateTime?>(null);
     }
 
     /// <summary>
@@ -283,6 +289,9 @@ public class KpiHistoryRecorderActorTests : TestKit
 
         public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
             Task.CompletedTask;
+
+        public Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
+            Task.FromResult<DateTime?>(null);
     }
 
     /// <summary>
@@ -340,6 +349,9 @@ public class KpiHistoryRecorderActorTests : TestKit
 
         public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
             Task.CompletedTask;
+
+        public Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
+            Task.FromResult<DateTime?>(null);
     }
 
     private IServiceProvider BuildServiceProvider(