diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs index 3aee78f3..e0070b92 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs @@ -76,12 +76,18 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService { var effectiveMax = maxPoints ?? _options.DefaultMaxSeriesPoints; + // Per-metric reduction intent: the same catalog that drives the hourly fold drives + // the chart-time bucket reduction, so a Rate series is summed per bucket on BOTH the + // raw-minute and hourly-rollup routes — one unit ("events per chart bucket") on both + // sides of the RollupThresholdHours routing boundary (arch-review 04 round 2, R3). + var aggregation = KpiMetricAggregationCatalog.Resolve(source, metric); + // Test-seam ctor: use the injected repository directly. if (_injectedRepository is not null) { var injectedSeries = await FetchSeriesAsync( _injectedRepository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken); - return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax); + return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax, aggregation); } // Production: a fresh scope (and thus a fresh DbContext) per query so a @@ -90,7 +96,7 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService var repository = serviceScope.ServiceProvider.GetRequiredService(); var series = await FetchSeriesAsync( repository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken); - return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax); + return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax, aggregation); } /// @@ -103,6 +109,13 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService /// 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. + /// + /// The routing boundary no longer changes a Rate chart's magnitude: both paths reduce to + /// per-bucket sums via the catalog-driven aggregation applied in + /// . A residual ≤~1.2× native-resolution difference exists only + /// for an unbucketed just-over-threshold rollup series (plotted at native per-hour sums) vs. + /// an at-threshold bucketed raw series — documented, not hidden (arch-review 04 round 2, R3). + /// /// private Task> FetchSeriesAsync( IKpiHistoryRepository repository, 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 0dbc6b2c..4a45e5bd 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs @@ -330,4 +330,95 @@ public class KpiHistoryQueryServiceTests Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); } + + // ---- Task 6: catalog-driven aggregation at the bucketer boundary (R3) ----------- + + [Fact] + public async Task GetSeriesAsync_RateMetric_SumsPerBucket_OnRollupPath() + { + // (SiteCallAudit, FailedLastInterval) is a catalog Rate pair. 400 hourly points of + // value 1 over a 400 h (>168 h) window, maxPoints 200 → 2 points per 2 h bucket → + // every output value is 2.0 (the summed pair), not 1.0 (last value). + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + var to = from.AddHours(400); + var rollup = Enumerable.Range(0, 400) + .Select(i => new KpiSeriesPoint(from.AddHours(i), 1d)) + .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: 200)); + + var result = await sut.GetSeriesAsync( + KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.FailedLastInterval, + "global", null, from, to); + + Assert.Equal(200, result.Count); + Assert.All(result, p => Assert.Equal(2.0, p.Value)); + } + + [Fact] + public async Task GetSeriesAsync_RateMetric_PreservesTotal_AcrossRoutingBoundary() + { + // Same total event count (50) served as per-minute deltas on the raw path + // (window == threshold) and as hourly sums on the rollup path (window just over + // threshold): the summed chart total must be equal on both routes — the boundary + // is visually seamless for a Rate metric. + var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + + var rawDeltas = Enumerable.Range(0, 10) + .Select(i => new KpiSeriesPoint(from.AddMinutes(i * 10), 5d)).ToList(); // total 50 + var hourlySums = Enumerable.Range(0, 5) + .Select(i => new KpiSeriesPoint(from.AddHours(i), 10d)).ToList(); // total 50 + + var repo = Substitute.For(); + repo.GetRawSeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(rawDeltas)); + repo.GetHourlySeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(hourlySums)); + + var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200, rollupThresholdHours: 168)); + + var rawResult = await sut.GetSeriesAsync( + KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.FailedLastInterval, + "global", null, from, from.AddHours(168)); // raw path + var rollupResult = await sut.GetSeriesAsync( + KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.FailedLastInterval, + "global", null, from, from.AddHours(169)); // rollup path + + Assert.Equal(50d, rawResult.Sum(p => p.Value)); + Assert.Equal(50d, rollupResult.Sum(p => p.Value)); + Assert.Equal(rawResult.Sum(p => p.Value), rollupResult.Sum(p => p.Value)); + } + + [Fact] + public async Task GetSeriesAsync_GaugeMetric_KeepsLastValuePerBucket() + { + // queueDepth (uncatalogued for this source pairing → Gauge) keeps the existing + // last-value-per-bucket output unchanged. 4 points, maxPoints 2, raw path. + var repo = Substitute.For(); + var raw = Series((0, 1d), (15, 2d), (25, 99d), (35, 5d)); + repo.GetRawSeriesAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromResult(raw)); + + var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 2)); + + var result = await sut.GetSeriesAsync( + KpiSources.NotificationOutbox, KpiMetrics.NotificationOutbox.QueueDepth, + "global", null, From, To); + + Assert.Equal(2, result.Count); + Assert.Equal(99d, result[0].Value); // last value in bucket 0, not a sum + Assert.Equal(5d, result[1].Value); + } }