diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs
index f486baab..1ebb8a78 100644
--- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/IKpiHistoryRepository.cs
@@ -50,4 +50,75 @@ public interface IKpiHistoryRepository
/// Cancellation token.
/// A task resolving to the number of rows deleted.
Task PurgeOlderThanAsync(DateTime before, CancellationToken cancellationToken = default);
+
+ ///
+ /// Folds the raw KpiSample rows in the window into the hourly
+ /// KpiRollupHourly table — one row per
+ /// (Source, Metric, Scope, ScopeKey) per UTC hour.
+ /// The window is truncated to whole-hour boundaries: only the complete hours in
+ /// [truncate(fromHourUtc), truncate(toHourUtc)) are folded — the upper bound
+ /// is exclusive, so a caller that passes the current hour's start as
+ /// never folds the in-progress (incomplete) hour.
+ ///
+ ///
+ ///
+ /// For each (series, hour) group the fold records SampleCount (raw rows in the
+ /// group), MinValue/MaxValue (min/max raw value), and a Value
+ /// whose aggregation is chosen per metric by
+ /// :
+ /// stores the
+ /// last (latest-timestamp) value in the hour;
+ /// stores the sum of values in the hour.
+ ///
+ ///
+ /// The write is an idempotent upsert keyed on
+ /// (Source, Metric, Scope, ScopeKey, HourStartUtc)
+ /// with null ScopeKey participating in equality: an existing row is updated
+ /// in place, otherwise a new row is inserted. Re-running the fold over the same window
+ /// yields identical rows (no duplicates, no double-counting), so a singleton-failover
+ /// handover that missed a tick self-heals on the next re-fold. Metrics that are only
+ /// conditionally emitted (missing in some minutes) are handled naturally by grouping.
+ ///
+ ///
+ /// Inclusive lower bound of the fold window (UTC), truncated to the hour.
+ /// Exclusive upper bound of the fold window (UTC), truncated to the hour; pass the current hour's start to exclude the in-progress hour.
+ /// Cancellation token.
+ /// A task that represents the asynchronous fold + upsert.
+ Task FoldHourlyRollupsAsync(
+ DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default);
+
+ ///
+ /// Returns the hourly rollup points for one series in [fromUtc, toUtc], ordered by
+ ///
+ /// ascending. Same contract and return shape as
+ /// (each point carries the hour start plus the folded
+ /// Value) so the downstream bucketer is agnostic to raw-vs-rollup source.
+ ///
+ /// Source identifier — a value from .
+ /// Metric name from the source's catalog.
+ /// Scope discriminator — a value from .
+ /// Scope qualifier (site id / node name); null for the Global scope.
+ /// Inclusive lower bound of the time window (UTC).
+ /// Inclusive upper bound of the time window (UTC).
+ /// Cancellation token.
+ ///
+ /// A task resolving to the hourly series points in ascending hour order; empty when the
+ /// series has no rollups in the window.
+ ///
+ Task> GetHourlySeriesAsync(
+ string source, string metric, string scope, string? scopeKey,
+ DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default);
+
+ ///
+ /// Bulk-deletes hourly rollup rows whose
+ /// is
+ /// strictly older than (rollup retention purge). Mirrors
+ /// : the purge is sliced into one-hour DELETE batches
+ /// so a large catch-up never runs as a single lock/log-heavy transaction
+ /// (arch-review 04, P4).
+ ///
+ /// UTC cut-off; rollups whose hour start is before this instant are deleted.
+ /// Cancellation token.
+ /// A task that represents the asynchronous purge.
+ Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default);
}
diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs
index 421e764f..d772b543 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs
@@ -82,4 +82,134 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
return total;
}
+
+ ///
+ public async Task FoldHourlyRollupsAsync(
+ DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default)
+ {
+ // Truncate the window to whole-hour boundaries. The fold covers the complete hours
+ // in [truncate(from), truncate(to)) — the upper bound is EXCLUSIVE, so a caller that
+ // passes the current hour's start as toHourUtc never folds the in-progress hour.
+ var from = TruncateToHour(fromHourUtc);
+ var to = TruncateToHour(toHourUtc);
+ if (to <= from)
+ {
+ return;
+ }
+
+ // The caller passes a small trailing lookback (e.g. 3 h), so the window is bounded:
+ // fetch it and group IN MEMORY. This deliberately avoids provider-specific
+ // DATEADD/DATEPART hour-truncation translation (SQLite in tests, SQL Server in prod),
+ // keeping the hour-bucket arithmetic identical across providers and trivially correct.
+ var samples = await _context.KpiSamples
+ .Where(s => s.CapturedAtUtc >= from && s.CapturedAtUtc < to)
+ .ToListAsync(cancellationToken);
+ if (samples.Count == 0)
+ {
+ return;
+ }
+
+ var groups = samples.GroupBy(s => new SeriesHourKey(
+ s.Source, s.Metric, s.Scope, s.ScopeKey, TruncateToHour(s.CapturedAtUtc)));
+
+ foreach (var group in groups)
+ {
+ var key = group.Key;
+
+ // Per-metric aggregation intent: Rate metrics sum the hour (last-value would
+ // discard 59 of 60 intra-hour deltas); Gauge metrics keep the last (latest
+ // timestamp) value in the hour. Min/Max/Count preserve the fold's fidelity.
+ var aggregation = KpiMetricAggregationCatalog.Resolve(key.Source, key.Metric);
+ var value = aggregation == KpiRollupAggregation.Rate
+ ? group.Sum(s => s.Value)
+ : group.OrderByDescending(s => s.CapturedAtUtc).First().Value;
+ var minValue = group.Min(s => s.Value);
+ var maxValue = group.Max(s => s.Value);
+ var sampleCount = group.Count();
+
+ // Idempotent upsert on the unique series+hour key. The ScopeKey == key.ScopeKey
+ // comparison matches null against the Global-scope rows (IS NULL) exactly as the
+ // UNIQUE IX_KpiRollupHourly_Series index treats a null key as participating.
+ var existing = await _context.KpiRollupHourly.FirstOrDefaultAsync(
+ r => r.Source == key.Source
+ && r.Metric == key.Metric
+ && r.Scope == key.Scope
+ && r.ScopeKey == key.ScopeKey
+ && r.HourStartUtc == key.HourStartUtc,
+ cancellationToken);
+
+ if (existing is null)
+ {
+ _context.KpiRollupHourly.Add(new KpiRollupHourly
+ {
+ Source = key.Source,
+ Metric = key.Metric,
+ Scope = key.Scope,
+ ScopeKey = key.ScopeKey,
+ HourStartUtc = key.HourStartUtc,
+ Value = value,
+ MinValue = minValue,
+ MaxValue = maxValue,
+ SampleCount = sampleCount,
+ });
+ }
+ else
+ {
+ // Re-fold overwrites the aggregate in place — a re-run over the same window
+ // produces identical values (no double-count), so a missed tick self-heals.
+ existing.Value = value;
+ existing.MinValue = minValue;
+ existing.MaxValue = maxValue;
+ existing.SampleCount = sampleCount;
+ }
+ }
+
+ await _context.SaveChangesAsync(cancellationToken);
+ }
+
+ ///
+ public async Task> GetHourlySeriesAsync(
+ string source, string metric, string scope, string? scopeKey,
+ DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default)
+ {
+ // Same null-ScopeKey semantics as GetRawSeriesAsync: scopeKey == null translates to
+ // "ScopeKey IS NULL", matching Global-scope rollups and excluding site/node-keyed ones.
+ return await _context.KpiRollupHourly
+ .Where(r => r.Source == source
+ && r.Metric == metric
+ && r.Scope == scope
+ && r.ScopeKey == scopeKey
+ && r.HourStartUtc >= fromUtc
+ && r.HourStartUtc <= toUtc)
+ .OrderBy(r => r.HourStartUtc)
+ .Select(r => new KpiSeriesPoint(r.HourStartUtc, r.Value))
+ .ToListAsync(cancellationToken);
+ }
+
+ ///
+ public async Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default)
+ {
+ // Mirror PurgeOlderThanAsync: one-hour-sliced batched DELETEs cap the lock/log
+ // footprint per statement (arch-review 04, P4). Rollups are one-per-hour, so each
+ // slice deletes a single hour bucket.
+ var floor = await _context.KpiRollupHourly.Where(r => r.HourStartUtc < before)
+ .MinAsync(r => (DateTime?)r.HourStartUtc, cancellationToken);
+ while (floor is not null && floor < before)
+ {
+ var ceiling = floor.Value.AddHours(1) < before ? floor.Value.AddHours(1) : before;
+ await _context.KpiRollupHourly
+ .Where(r => r.HourStartUtc < ceiling)
+ .ExecuteDeleteAsync(cancellationToken);
+ floor = await _context.KpiRollupHourly.Where(r => r.HourStartUtc < before)
+ .MinAsync(r => (DateTime?)r.HourStartUtc, cancellationToken);
+ }
+ }
+
+ /// Truncates a UTC timestamp to the start of its hour (minutes/seconds/ticks zeroed).
+ private static DateTime TruncateToHour(DateTime utc) =>
+ new(utc.Year, utc.Month, utc.Day, utc.Hour, 0, 0, DateTimeKind.Utc);
+
+ /// In-memory grouping key: one KPI series (four-tuple) within one UTC hour.
+ private readonly record struct SeriesHourKey(
+ string Source, string Metric, string Scope, string? ScopeKey, DateTime HourStartUtc);
}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs
index d7323e25..68822e4e 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs
@@ -1,4 +1,5 @@
using System.Data.Common;
+using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Diagnostics;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
@@ -195,6 +196,200 @@ public class KpiHistoryRepositoryTests
Assert.Equal(new[] { 4d, 5d }, remaining.Select(p => p.Value).ToArray());
}
+ // ---- Hourly rollup fold / query / purge (plan #22 T3) ------------------------------
+
+ [Fact]
+ public async Task FoldHourlyRollupsAsync_GaugeMetric_FoldsToLastValuePerHour()
+ {
+ await using var ctx = NewContext();
+ var repo = new KpiHistoryRepository(ctx);
+
+ // "queueDepth" is a Gauge → last (latest-timestamp) value in the hour wins.
+ await repo.RecordSamplesAsync(new[]
+ {
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(5)),
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 9, capturedAtUtc: Base.AddMinutes(59)),
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 7, capturedAtUtc: Base.AddMinutes(30)),
+ });
+
+ // toHourUtc = Base + 1 h (next hour start) → the Base hour is complete and folded.
+ await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
+
+ var series = await repo.GetHourlySeriesAsync(
+ "NotificationOutbox", "queueDepth", "Global", scopeKey: null,
+ fromUtc: Base, toUtc: Base.AddHours(24));
+
+ var row = Assert.Single(series);
+ Assert.Equal(Base, row.BucketStartUtc);
+ Assert.Equal(9, row.Value); // last-value (max CapturedAtUtc)
+ }
+
+ [Fact]
+ public async Task FoldHourlyRollupsAsync_RateMetric_FoldsToSumPerHour_WithMinMaxCount()
+ {
+ await using var ctx = NewContext();
+ var repo = new KpiHistoryRepository(ctx);
+
+ // "deliveredLastInterval" is a Rate → sum of the hour's per-interval deltas.
+ await repo.RecordSamplesAsync(new[]
+ {
+ Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 2, capturedAtUtc: Base.AddMinutes(5)),
+ Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 5, capturedAtUtc: Base.AddMinutes(25)),
+ Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 4, capturedAtUtc: Base.AddMinutes(45)),
+ });
+
+ await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
+
+ // Read the entity directly to assert Min/Max/Count fidelity, not just Value.
+ var rollup = Assert.Single(await ctx.KpiRollupHourly.ToListAsync());
+ Assert.Equal(Base, rollup.HourStartUtc);
+ Assert.Equal(11, rollup.Value); // sum 2+5+4
+ Assert.Equal(2, rollup.MinValue);
+ Assert.Equal(5, rollup.MaxValue);
+ Assert.Equal(3, rollup.SampleCount);
+ }
+
+ [Fact]
+ public async Task FoldHourlyRollupsAsync_SplitsIntoSeparateHourBuckets_ForGlobalScope()
+ {
+ await using var ctx = NewContext();
+ var repo = new KpiHistoryRepository(ctx);
+
+ // Two hours of one Global (null ScopeKey) gauge series.
+ await repo.RecordSamplesAsync(new[]
+ {
+ Sample("SiteHealth", "connectionsUp", "Global", null, value: 4, capturedAtUtc: Base.AddMinutes(10)),
+ Sample("SiteHealth", "connectionsUp", "Global", null, value: 6, capturedAtUtc: Base.AddMinutes(50)),
+ Sample("SiteHealth", "connectionsUp", "Global", null, value: 2, capturedAtUtc: Base.AddHours(1).AddMinutes(20)),
+ });
+
+ // Fold two complete hours: [Base, Base+2h).
+ await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(2));
+
+ var series = await repo.GetHourlySeriesAsync(
+ "SiteHealth", "connectionsUp", "Global", scopeKey: null,
+ fromUtc: Base, toUtc: Base.AddHours(24));
+
+ Assert.Equal(2, series.Count);
+ Assert.Equal(Base, series[0].BucketStartUtc);
+ Assert.Equal(6, series[0].Value); // hour 0 last-value
+ Assert.Equal(Base.AddHours(1), series[1].BucketStartUtc);
+ Assert.Equal(2, series[1].Value); // hour 1 last-value
+ }
+
+ [Fact]
+ public async Task FoldHourlyRollupsAsync_ExcludesIncompleteHour_AtExclusiveUpperBound()
+ {
+ await using var ctx = NewContext();
+ var repo = new KpiHistoryRepository(ctx);
+
+ await repo.RecordSamplesAsync(new[]
+ {
+ // Complete hour (Base) — folded.
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(30)),
+ // Next, in-progress hour — must NOT be folded when toHourUtc == Base+1h.
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 99, capturedAtUtc: Base.AddHours(1).AddMinutes(15)),
+ });
+
+ await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
+
+ var series = await repo.GetHourlySeriesAsync(
+ "NotificationOutbox", "queueDepth", "Global", scopeKey: null,
+ fromUtc: Base, toUtc: Base.AddHours(24));
+
+ // Only the complete Base hour is present; the in-progress hour is excluded.
+ var row = Assert.Single(series);
+ Assert.Equal(Base, row.BucketStartUtc);
+ Assert.Equal(3, row.Value);
+ }
+
+ [Fact]
+ public async Task FoldHourlyRollupsAsync_IsIdempotent_AcrossRepeatedRunsOverSameWindow()
+ {
+ await using var ctx = NewContext();
+ var repo = new KpiHistoryRepository(ctx);
+
+ await repo.RecordSamplesAsync(new[]
+ {
+ Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 2, capturedAtUtc: Base.AddMinutes(5)),
+ Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 5, capturedAtUtc: Base.AddMinutes(45)),
+ });
+
+ await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
+ await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1)); // re-run same window
+
+ // Exactly one row for the (series, hour); the rate sum is not double-counted.
+ var rollups = await ctx.KpiRollupHourly.ToListAsync();
+ var row = Assert.Single(rollups);
+ Assert.Equal(7, row.Value); // still 2+5, not 14
+ Assert.Equal(2, row.SampleCount);
+ }
+
+ [Fact]
+ public async Task GetHourlySeriesAsync_ReturnsAscending_AndHonorsNullVsSiteScopeKey()
+ {
+ await using var ctx = NewContext();
+ var repo = new KpiHistoryRepository(ctx);
+
+ // Same metric, one Global (null key) hour and one Site-keyed hour, both foldable.
+ await repo.RecordSamplesAsync(new[]
+ {
+ Sample("SiteCallAudit", "buffered", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(10)),
+ Sample("SiteCallAudit", "buffered", "Global", null, value: 8, capturedAtUtc: Base.AddHours(1).AddMinutes(10)),
+ Sample("SiteCallAudit", "buffered", "Site", "plant-a", value: 42, capturedAtUtc: Base.AddMinutes(10)),
+ });
+
+ await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(2));
+
+ var global = await repo.GetHourlySeriesAsync(
+ "SiteCallAudit", "buffered", "Global", scopeKey: null,
+ fromUtc: Base, toUtc: Base.AddHours(24));
+
+ // Two Global hours, ascending; the Site-keyed row must not leak in.
+ Assert.Equal(2, global.Count);
+ Assert.Equal(Base, global[0].BucketStartUtc);
+ Assert.Equal(3, global[0].Value);
+ Assert.Equal(Base.AddHours(1), global[1].BucketStartUtc);
+ Assert.Equal(8, global[1].Value);
+
+ var site = await repo.GetHourlySeriesAsync(
+ "SiteCallAudit", "buffered", "Site", scopeKey: "plant-a",
+ fromUtc: Base, toUtc: Base.AddHours(24));
+ Assert.Equal(42, Assert.Single(site).Value);
+ }
+
+ [Fact]
+ public async Task PurgeRollupsOlderThanAsync_DeletesOnlyRollupsOlderThanCutoff()
+ {
+ await using var ctx = NewContext();
+ var repo = new KpiHistoryRepository(ctx);
+
+ // Fold four distinct hours spanning several days.
+ await repo.RecordSamplesAsync(new[]
+ {
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 1, capturedAtUtc: Base.AddDays(-10).AddMinutes(5)),
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 2, capturedAtUtc: Base.AddDays(-8).AddMinutes(5)),
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddDays(-7).AddMinutes(5)),
+ Sample("NotificationOutbox", "queueDepth", "Global", null, value: 4, capturedAtUtc: Base.AddDays(-1).AddMinutes(5)),
+ });
+ await repo.FoldHourlyRollupsAsync(Base.AddDays(-30), Base);
+
+ // Cutoff = the -7d hour start; strictly-older predicate keeps the ==cutoff row.
+ var cutoff = TruncateHour(Base.AddDays(-7));
+ await repo.PurgeRollupsOlderThanAsync(cutoff);
+
+ var remaining = await repo.GetHourlySeriesAsync(
+ "NotificationOutbox", "queueDepth", "Global", scopeKey: null,
+ fromUtc: Base.AddDays(-30), toUtc: Base.AddDays(30));
+
+ // The -7d (==cutoff) and -1d rollups survive; the -10d and -8d are purged.
+ Assert.Equal(new[] { 3d, 4d }, remaining.Select(p => p.Value).ToArray());
+ }
+
+ // 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);
+
///
/// EF command interceptor that counts the DELETE statements actually issued
/// to the store, so a test can prove the purge is sliced into multiple