From a10170f365814f6f980db0f7259843eade53e923 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 10 Jul 2026 12:01:59 -0400 Subject: [PATCH] feat(kpihistory): route long-range KPI queries to hourly rollups by threshold (plan #22 T5) GetSeriesAsync now branches on window width: windows wider than RollupThresholdHours (default 168h/7d) read pre-aggregated hourly rollups via GetHourlySeriesAsync; windows at or below the threshold read raw samples via GetRawSeriesAsync (preserving intra-minute detail on 24h/7d). The boundary is strict greater-than, so exactly 168h stays on the raw path. KpiSeriesBucketer.Bucket runs unchanged on whichever series (a 90d rollup can still exceed the point ceiling). Both ctors route identically via a shared FetchSeriesAsync helper reading _options.RollupThresholdHours, the same options instance that already supplies DefaultMaxSeriesPoints. GetSeriesAsync public signature unchanged. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj --- .../Services/KpiHistoryQueryService.cs | 34 +++- .../Services/KpiHistoryQueryServiceTests.cs | 158 +++++++++++++++++- 2 files changed, 184 insertions(+), 8 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs index aa69346f..3aee78f3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs @@ -79,17 +79,39 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService // Test-seam ctor: use the injected repository directly. if (_injectedRepository is not null) { - var injectedRaw = await _injectedRepository.GetRawSeriesAsync( - source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken); - return KpiSeriesBucketer.Bucket(injectedRaw, fromUtc, toUtc, effectiveMax); + var injectedSeries = await FetchSeriesAsync( + _injectedRepository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken); + return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax); } // Production: a fresh scope (and thus a fresh DbContext) per query so a // chart's auto-load never shares the circuit-scoped context. await using var serviceScope = _scopeFactory!.CreateAsyncScope(); var repository = serviceScope.ServiceProvider.GetRequiredService(); - var raw = await repository.GetRawSeriesAsync( - source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken); - return KpiSeriesBucketer.Bucket(raw, fromUtc, toUtc, effectiveMax); + var series = await FetchSeriesAsync( + repository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken); + return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax); + } + + /// + /// Routes a series fetch to the raw-sample or hourly-rollup repository method by + /// the requested window width. Windows wider than + /// read the pre-aggregated + /// hourly rollups (≈60× fewer rows scanned on 30 d/90 d ranges); windows at or below + /// the threshold read raw samples, preserving intra-minute detail on the 24 h/7 d + /// charts. Both paths return the same ascending shape, so + /// the caller's step is agnostic to the source + /// (a 90 d rollup can still exceed the point ceiling, so bucketing still applies). + /// The boundary is strict: exactly RollupThresholdHours stays on the raw path. + /// + private Task> FetchSeriesAsync( + IKpiHistoryRepository repository, + string source, string metric, string scope, string? scopeKey, + DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken) + { + var windowHours = (toUtc - fromUtc).TotalHours; + return windowHours > _options.RollupThresholdHours + ? repository.GetHourlySeriesAsync(source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken) + : repository.GetRawSeriesAsync(source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs index 8d32e129..0dbc6b2c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs @@ -20,8 +20,12 @@ public class KpiHistoryQueryServiceTests private static readonly DateTime From = new(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc); private static readonly DateTime To = new(2026, 5, 20, 11, 0, 0, DateTimeKind.Utc); - private static IOptions Options(int defaultMax) => - Microsoft.Extensions.Options.Options.Create(new KpiHistoryOptions { DefaultMaxSeriesPoints = defaultMax }); + private static IOptions Options(int defaultMax, int rollupThresholdHours = 168) => + Microsoft.Extensions.Options.Options.Create(new KpiHistoryOptions + { + DefaultMaxSeriesPoints = defaultMax, + RollupThresholdHours = rollupThresholdHours, + }); private static IReadOnlyList Series(params (int minute, double value)[] pts) => pts.Select(p => new KpiSeriesPoint(From.AddMinutes(p.minute), p.value)).ToList(); @@ -176,4 +180,154 @@ public class KpiHistoryQueryServiceTests Assert.Equal(2, resolvedRepos.Count); Assert.NotSame(resolvedRepos[0], resolvedRepos[1]); } + + // ---- Task 5: raw-vs-rollup range routing by RollupThresholdHours ---------------- + + private static IKpiHistoryRepository RepoReturningEmptyForBoth() + { + var repo = Substitute.For(); + repo.GetRawSeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(Array.Empty())); + repo.GetHourlySeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(Array.Empty())); + return repo; + } + + [Fact] + public async Task GetSeriesAsync_24hWindow_UsesRawPath() + { + var repo = RepoReturningEmptyForBoth(); + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + var to = from.AddHours(24); + + var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200)); + + await sut.GetSeriesAsync("src", "metric", "global", null, from, to); + + await repo.Received(1).GetRawSeriesAsync( + "src", "metric", "global", null, from, to, Arg.Any()); + await repo.DidNotReceive().GetHourlySeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task GetSeriesAsync_WindowExactlyAtThreshold_UsesRawPath() + { + // Boundary is strictly greater-than: exactly 168 h stays on the raw path. + var repo = RepoReturningEmptyForBoth(); + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + var to = from.AddHours(168); + + var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200, rollupThresholdHours: 168)); + + await sut.GetSeriesAsync("src", "metric", "global", null, from, to); + + await repo.Received(1).GetRawSeriesAsync( + "src", "metric", "global", null, from, to, Arg.Any()); + await repo.DidNotReceive().GetHourlySeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task GetSeriesAsync_WindowJustOverThreshold_UsesRollupPath() + { + // 169 h (threshold + 1) crosses the boundary → hourly rollup path. + var repo = RepoReturningEmptyForBoth(); + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + var to = from.AddHours(169); + + var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200, rollupThresholdHours: 168)); + + await sut.GetSeriesAsync("src", "metric", "global", null, from, to); + + await repo.Received(1).GetHourlySeriesAsync( + "src", "metric", "global", null, from, to, Arg.Any()); + await repo.DidNotReceive().GetRawSeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task GetSeriesAsync_90dWindow_UsesRollupPath() + { + var repo = RepoReturningEmptyForBoth(); + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + var to = from.AddDays(90); + + var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200)); + + await sut.GetSeriesAsync("src", "metric", "global", null, from, to); + + await repo.Received(1).GetHourlySeriesAsync( + "src", "metric", "global", null, from, to, Arg.Any()); + await repo.DidNotReceive().GetRawSeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task GetSeriesAsync_RollupPath_StillBucketsWhenOverMax() + { + // A rollup series longer than maxPoints must still be downsampled by the + // bucketer on top of the rollup read — the bucketer path is unchanged. + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + var to = from.AddDays(90); + + // 6 hourly rollup points, spread across the window in ascending order. + var rollup = Enumerable.Range(0, 6) + .Select(i => new KpiSeriesPoint(from.AddHours(i * 300), i)) + .ToList(); + + var repo = Substitute.For(); + repo.GetHourlySeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(rollup)); + + var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 3)); + + var result = await sut.GetSeriesAsync("src", "metric", "global", null, from, to); + + // Routed to the rollup path... + await repo.Received(1).GetHourlySeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()); + // ...and the bucketer still applied (6 rollup rows > max of 3 → downsampled). + Assert.NotSame(rollup, result); + Assert.True(result.Count <= 3); + } + + [Fact] + public async Task GetSeriesAsync_LongWindow_UsesRollupPath_OnProductionCtor() + { + // The production (IServiceScopeFactory) ctor must apply the same threshold + // routing — a long window resolves the repo from a fresh scope and calls the + // hourly rollup method. + var repo = RepoReturningEmptyForBoth(); + + var services = new ServiceCollection(); + services.AddScoped(_ => repo); + + await using var provider = services.BuildServiceProvider(); + var sut = new KpiHistoryQueryService( + provider.GetRequiredService(), + Options(defaultMax: 200)); + + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + var to = from.AddDays(30); + + await sut.GetSeriesAsync("src", "metric", "global", null, from, to); + + await repo.Received(1).GetHourlySeriesAsync( + "src", "metric", "global", null, from, to, Arg.Any()); + await repo.DidNotReceive().GetRawSeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()); + } }