feat(kpi): K7 — SiteCallAudit sample source

src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/Kpi/SiteCallAuditKpiSampleSource.cs

@@ -0,0 +1,159 @@
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
+
+namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit.Kpi;
+
+/// <summary>
+/// Site Call Audit (#22) <see cref="IKpiSampleSource"/> for the M6 "KPI History
+/// &amp; Trends" backbone. Each sampling pass the central recorder enumerates this
+/// source and calls <see cref="CollectAsync"/>, which snapshots the same
+/// point-in-time <c>SiteCalls</c> KPIs the live Health-dashboard tiles show —
+/// global, per-source-site, and per-originating-node — into flat
+/// <see cref="KpiSample"/> rows.
+/// </summary>
+/// <remarks>
+/// <para>
+/// The cutoffs are derived from <see cref="SiteCallAuditOptions"/> exactly as the
+/// live <c>SiteCallAuditActor</c> KPI handlers derive them
+/// (<c>stuckCutoff = capturedAtUtc - StuckAgeThreshold</c>,
+/// <c>intervalSince = capturedAtUtc - KpiInterval</c>) so a sampled value equals
+/// the live tile computed at the same instant. The recorder's
+/// <c>capturedAtUtc</c> is the single anchor for both cutoffs.
+/// </para>
+/// <para>
+/// Registered DI-scoped (next to the rest of the Site Call Audit composition) so
+/// each sampling pass resolves a fresh repository scope, mirroring the actor's
+/// scope-per-message repository access.
+/// </para>
+/// </remarks>
+public sealed class SiteCallAuditKpiSampleSource : IKpiSampleSource
+{
+    // ── Metric catalog (the M6-agreed metric-name strings for this source) ──
+    private const string MetricBuffered = "buffered";
+    private const string MetricParked = "parked";
+    private const string MetricFailedLastInterval = "failedLastInterval";
+    private const string MetricDeliveredLastInterval = "deliveredLastInterval";
+    private const string MetricOldestPendingAgeSeconds = "oldestPendingAgeSeconds";
+    private const string MetricStuck = "stuck";
+
+    private readonly ISiteCallAuditRepository _repository;
+    private readonly SiteCallAuditOptions _options;
+
+    /// <summary>
+    /// Creates the sample source over the central <c>SiteCalls</c> repository and
+    /// the Site Call Audit options that define the stuck-age + KPI-interval
+    /// windows.
+    /// </summary>
+    /// <param name="repository">The central <c>SiteCalls</c> operational-state repository.</param>
+    /// <param name="options">Site Call Audit windowing options (stuck-age + KPI interval).</param>
+    public SiteCallAuditKpiSampleSource(
+        ISiteCallAuditRepository repository,
+        IOptions<SiteCallAuditOptions> options)
+    {
+        ArgumentNullException.ThrowIfNull(repository);
+        ArgumentNullException.ThrowIfNull(options);
+
+        _repository = repository;
+        _options = options.Value;
+    }
+
+    /// <inheritdoc />
+    public string Source => KpiSources.SiteCallAudit;
+
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<KpiSample>> CollectAsync(
+        DateTime capturedAtUtc, CancellationToken cancellationToken = default)
+    {
+        // Match the live SiteCallAuditActor KPI handlers: stuck cutoff and
+        // interval window are both anchored on the single capture instant.
+        var stuckCutoff = capturedAtUtc - _options.StuckAgeThreshold;
+        var intervalSince = capturedAtUtc - _options.KpiInterval;
+
+        var global = await _repository
+            .ComputeKpisAsync(stuckCutoff, intervalSince, cancellationToken)
+            .ConfigureAwait(false);
+        var perSite = await _repository
+            .ComputePerSiteKpisAsync(stuckCutoff, intervalSince, cancellationToken)
+            .ConfigureAwait(false);
+        var perNode = await _repository
+            .ComputePerNodeKpisAsync(stuckCutoff, intervalSince, cancellationToken)
+            .ConfigureAwait(false);
+
+        var samples = new List<KpiSample>();
+
+        // Global scope (null ScopeKey).
+        AddSnapshot(
+            samples, capturedAtUtc, KpiScopes.Global, scopeKey: null,
+            global.BufferedCount, global.ParkedCount,
+            global.FailedLastInterval, global.DeliveredLastInterval,
+            global.OldestPendingAge, global.StuckCount);
+
+        // Per-site scope (ScopeKey = source site id).
+        foreach (var site in perSite)
+        {
+            AddSnapshot(
+                samples, capturedAtUtc, KpiScopes.Site, scopeKey: site.SourceSite,
+                site.BufferedCount, site.ParkedCount,
+                site.FailedLastInterval, site.DeliveredLastInterval,
+                site.OldestPendingAge, site.StuckCount);
+        }
+
+        // Per-node scope (ScopeKey = node name).
+        foreach (var node in perNode)
+        {
+            AddSnapshot(
+                samples, capturedAtUtc, KpiScopes.Node, scopeKey: node.SourceNode,
+                node.BufferedCount, node.ParkedCount,
+                node.FailedLastInterval, node.DeliveredLastInterval,
+                node.OldestPendingAge, node.StuckCount);
+        }
+
+        return samples;
+    }
+
+    /// <summary>
+    /// Appends the six-metric catalog for one snapshot at the given scope. The
+    /// oldest-pending-age metric is omitted when the snapshot's age is
+    /// <c>null</c> (no non-terminal rows) rather than written as zero.
+    /// </summary>
+    private void AddSnapshot(
+        List<KpiSample> samples,
+        DateTime capturedAtUtc,
+        string scope,
+        string? scopeKey,
+        int buffered,
+        int parked,
+        int failedLastInterval,
+        int deliveredLastInterval,
+        TimeSpan? oldestPendingAge,
+        int stuck)
+    {
+        samples.Add(Sample(capturedAtUtc, MetricBuffered, scope, scopeKey, buffered));
+        samples.Add(Sample(capturedAtUtc, MetricParked, scope, scopeKey, parked));
+        samples.Add(Sample(capturedAtUtc, MetricFailedLastInterval, scope, scopeKey, failedLastInterval));
+        samples.Add(Sample(capturedAtUtc, MetricDeliveredLastInterval, scope, scopeKey, deliveredLastInterval));
+        samples.Add(Sample(capturedAtUtc, MetricStuck, scope, scopeKey, stuck));
+
+        if (oldestPendingAge is { } age)
+        {
+            samples.Add(Sample(
+                capturedAtUtc, MetricOldestPendingAgeSeconds, scope, scopeKey, age.TotalSeconds));
+        }
+    }
+
+    private KpiSample Sample(
+        DateTime capturedAtUtc, string metric, string scope, string? scopeKey, double value) =>
+        new()
+        {
+            Source = KpiSources.SiteCallAudit,
+            Metric = metric,
+            Scope = scope,
+            ScopeKey = scopeKey,
+            Value = value,
+            CapturedAtUtc = capturedAtUtc,
+        };
+}

src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/ServiceCollectionExtensions.cs

@@ -1,4 +1,6 @@
 using Microsoft.Extensions.DependencyInjection;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi;
+using ZB.MOM.WW.ScadaBridge.SiteCallAudit.Kpi;
 
 namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit;
 
@@ -37,6 +39,13 @@ public static class ServiceCollectionExtensions
         services.AddOptions<SiteCallAuditOptions>()
             .BindConfiguration(OptionsSection);
 
+        // M6 KPI History (#K7): the central recorder enumerates every registered
+        // IKpiSampleSource each sampling pass; this one snapshots the SiteCalls
+        // KPIs (global + per-site + per-node) into the KpiSample history store.
+        // Scoped so each pass resolves a fresh ISiteCallAuditRepository scope,
+        // mirroring the actor's scope-per-message repository access.
+        services.AddScoped<IKpiSampleSource, SiteCallAuditKpiSampleSource>();
+
         return services;
     }
 }

tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/Kpi/SiteCallAuditKpiSampleSourceTests.cs

@@ -0,0 +1,308 @@
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Commons.Types;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
+using ZB.MOM.WW.ScadaBridge.SiteCallAudit.Kpi;
+
+namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests.Kpi;
+
+/// <summary>
+/// Unit tests for <see cref="SiteCallAuditKpiSampleSource"/> (M6 KPI History,
+/// #K7). A hand-rolled <see cref="ISiteCallAuditRepository"/> fake (matching the
+/// project's existing test-double style — no mocking library) returns known
+/// global + per-site + per-node snapshots; the assertions pin the exact
+/// <c>(Metric, Scope, ScopeKey, Value)</c> tuples the recorder will persist, the
+/// cutoffs derived from <see cref="SiteCallAuditOptions"/>, and the
+/// oldest-pending-age omission when its source is <c>null</c>.
+/// </summary>
+public class SiteCallAuditKpiSampleSourceTests
+{
+    private static readonly DateTime CapturedAt =
+        new(2026, 6, 15, 12, 0, 0, DateTimeKind.Utc);
+
+    private static SiteCallAuditOptions Options(
+        TimeSpan? stuckAge = null, TimeSpan? kpiInterval = null) => new()
+    {
+        StuckAgeThreshold = stuckAge ?? TimeSpan.FromMinutes(10),
+        KpiInterval = kpiInterval ?? TimeSpan.FromMinutes(1),
+    };
+
+    private static SiteCallAuditKpiSampleSource CreateSource(
+        ISiteCallAuditRepository repo, SiteCallAuditOptions? options = null) =>
+        new(repo, OptionsWrap(options ?? Options()));
+
+    private static IOptions<SiteCallAuditOptions> OptionsWrap(SiteCallAuditOptions o) =>
+        Microsoft.Extensions.Options.Options.Create(o);
+
+    // ---------------------------------------------------------------------
+    // 1. Source identifier is the canonical SiteCallAudit constant.
+    // ---------------------------------------------------------------------
+
+    [Fact]
+    public void Source_IsSiteCallAuditConstant()
+    {
+        var source = CreateSource(new StubRepo());
+        Assert.Equal(KpiSources.SiteCallAudit, source.Source);
+    }
+
+    // ---------------------------------------------------------------------
+    // 2. Full snapshot: global + per-site + per-node, exact tuples.
+    // ---------------------------------------------------------------------
+
+    [Fact]
+    public async Task CollectAsync_EmitsExpectedTuples_ForGlobalSiteAndNode()
+    {
+        var repo = new StubRepo
+        {
+            Global = new SiteCallKpiSnapshot(
+                BufferedCount: 5,
+                ParkedCount: 2,
+                FailedLastInterval: 1,
+                DeliveredLastInterval: 9,
+                OldestPendingAge: TimeSpan.FromSeconds(42),
+                StuckCount: 3),
+            PerSite =
+            [
+                new SiteCallSiteKpiSnapshot(
+                    SourceSite: "site-a",
+                    BufferedCount: 4,
+                    ParkedCount: 1,
+                    FailedLastInterval: 0,
+                    DeliveredLastInterval: 7,
+                    OldestPendingAge: TimeSpan.FromSeconds(30),
+                    StuckCount: 2),
+            ],
+            PerNode =
+            [
+                new SiteCallNodeKpiSnapshot(
+                    SourceNode: "node-a",
+                    BufferedCount: 3,
+                    ParkedCount: 1,
+                    FailedLastInterval: 1,
+                    DeliveredLastInterval: 5,
+                    OldestPendingAge: TimeSpan.FromSeconds(20),
+                    StuckCount: 1),
+            ],
+        };
+
+        var samples = await CreateSource(repo).CollectAsync(CapturedAt);
+
+        // Every sample carries Source + CapturedAtUtc.
+        Assert.All(samples, s =>
+        {
+            Assert.Equal(KpiSources.SiteCallAudit, s.Source);
+            Assert.Equal(CapturedAt, s.CapturedAtUtc);
+        });
+
+        // Global (null ScopeKey): six metrics (age present).
+        AssertHas(samples, "buffered", KpiScopes.Global, null, 5);
+        AssertHas(samples, "parked", KpiScopes.Global, null, 2);
+        AssertHas(samples, "failedLastInterval", KpiScopes.Global, null, 1);
+        AssertHas(samples, "deliveredLastInterval", KpiScopes.Global, null, 9);
+        AssertHas(samples, "oldestPendingAgeSeconds", KpiScopes.Global, null, 42);
+        AssertHas(samples, "stuck", KpiScopes.Global, null, 3);
+
+        // Per-site (ScopeKey = site id).
+        AssertHas(samples, "buffered", KpiScopes.Site, "site-a", 4);
+        AssertHas(samples, "parked", KpiScopes.Site, "site-a", 1);
+        AssertHas(samples, "failedLastInterval", KpiScopes.Site, "site-a", 0);
+        AssertHas(samples, "deliveredLastInterval", KpiScopes.Site, "site-a", 7);
+        AssertHas(samples, "oldestPendingAgeSeconds", KpiScopes.Site, "site-a", 30);
+        AssertHas(samples, "stuck", KpiScopes.Site, "site-a", 2);
+
+        // Per-node (ScopeKey = node name).
+        AssertHas(samples, "buffered", KpiScopes.Node, "node-a", 3);
+        AssertHas(samples, "parked", KpiScopes.Node, "node-a", 1);
+        AssertHas(samples, "failedLastInterval", KpiScopes.Node, "node-a", 1);
+        AssertHas(samples, "deliveredLastInterval", KpiScopes.Node, "node-a", 5);
+        AssertHas(samples, "oldestPendingAgeSeconds", KpiScopes.Node, "node-a", 20);
+        AssertHas(samples, "stuck", KpiScopes.Node, "node-a", 1);
+
+        // 6 metrics × 3 scopes, all ages present.
+        Assert.Equal(18, samples.Count);
+    }
+
+    // ---------------------------------------------------------------------
+    // 3. Null oldest-pending-age is omitted (not written as zero).
+    // ---------------------------------------------------------------------
+
+    [Fact]
+    public async Task CollectAsync_OmitsOldestPendingAge_WhenNull()
+    {
+        var repo = new StubRepo
+        {
+            Global = new SiteCallKpiSnapshot(0, 0, 0, 0, OldestPendingAge: null, 0),
+            PerSite =
+            [
+                new SiteCallSiteKpiSnapshot("site-a", 0, 0, 0, 0, OldestPendingAge: null, 0),
+            ],
+            PerNode =
+            [
+                new SiteCallNodeKpiSnapshot("node-a", 0, 0, 0, 0, OldestPendingAge: null, 0),
+            ],
+        };
+
+        var samples = await CreateSource(repo).CollectAsync(CapturedAt);
+
+        // No oldestPendingAgeSeconds rows at any scope when the age is null.
+        Assert.DoesNotContain(samples, s => s.Metric == "oldestPendingAgeSeconds");
+
+        // The five count metrics are still present at each of the 3 scopes.
+        Assert.Equal(15, samples.Count);
+        AssertHas(samples, "buffered", KpiScopes.Global, null, 0);
+        AssertHas(samples, "stuck", KpiScopes.Node, "node-a", 0);
+    }
+
+    // ---------------------------------------------------------------------
+    // 4. Empty per-site/per-node: only the global six metrics emitted.
+    // ---------------------------------------------------------------------
+
+    [Fact]
+    public async Task CollectAsync_GlobalOnly_WhenNoPerSiteOrPerNodeRows()
+    {
+        var repo = new StubRepo
+        {
+            Global = new SiteCallKpiSnapshot(1, 0, 0, 0, TimeSpan.FromSeconds(5), 0),
+            PerSite = [],
+            PerNode = [],
+        };
+
+        var samples = await CreateSource(repo).CollectAsync(CapturedAt);
+
+        Assert.All(samples, s => Assert.Equal(KpiScopes.Global, s.Scope));
+        Assert.Equal(6, samples.Count);
+    }
+
+    // ---------------------------------------------------------------------
+    // 5. Cutoffs are anchored on capturedAtUtc using the options windows.
+    // ---------------------------------------------------------------------
+
+    [Fact]
+    public async Task CollectAsync_DerivesCutoffs_FromOptionsAnchoredOnCapturedAt()
+    {
+        var repo = new StubRepo();
+        var options = Options(
+            stuckAge: TimeSpan.FromMinutes(10),
+            kpiInterval: TimeSpan.FromMinutes(2));
+
+        await CreateSource(repo, options).CollectAsync(CapturedAt);
+
+        var expectedStuck = CapturedAt - TimeSpan.FromMinutes(10);
+        var expectedSince = CapturedAt - TimeSpan.FromMinutes(2);
+
+        // Every repository call must have received the identical anchored cutoffs.
+        Assert.NotEmpty(repo.Calls);
+        Assert.All(repo.Calls, c =>
+        {
+            Assert.Equal(expectedStuck, c.StuckCutoff);
+            Assert.Equal(expectedSince, c.IntervalSince);
+        });
+
+        // All three KPI methods were invoked.
+        Assert.Equal(3, repo.Calls.Count);
+    }
+
+    // ---------------------------------------------------------------------
+    // 6. Multiple sites/nodes are all emitted with distinct ScopeKeys.
+    // ---------------------------------------------------------------------
+
+    [Fact]
+    public async Task CollectAsync_EmitsAllSitesAndNodes()
+    {
+        var repo = new StubRepo
+        {
+            Global = new SiteCallKpiSnapshot(0, 0, 0, 0, null, 0),
+            PerSite =
+            [
+                new SiteCallSiteKpiSnapshot("site-a", 1, 0, 0, 0, null, 0),
+                new SiteCallSiteKpiSnapshot("site-b", 2, 0, 0, 0, null, 0),
+            ],
+            PerNode =
+            [
+                new SiteCallNodeKpiSnapshot("node-a", 0, 0, 0, 0, null, 0),
+                new SiteCallNodeKpiSnapshot("node-b", 0, 0, 0, 0, null, 0),
+            ],
+        };
+
+        var samples = await CreateSource(repo).CollectAsync(CapturedAt);
+
+        AssertHas(samples, "buffered", KpiScopes.Site, "site-a", 1);
+        AssertHas(samples, "buffered", KpiScopes.Site, "site-b", 2);
+        Assert.Contains(samples, s => s is { Scope: "Node", ScopeKey: "node-a" });
+        Assert.Contains(samples, s => s is { Scope: "Node", ScopeKey: "node-b" });
+    }
+
+    // ---------------------------------------------------------------------
+    // Helpers
+    // ---------------------------------------------------------------------
+
+    private static void AssertHas(
+        IReadOnlyList<KpiSample> samples,
+        string metric, string scope, string? scopeKey, double value)
+    {
+        var match = samples.SingleOrDefault(s =>
+            s.Metric == metric && s.Scope == scope && s.ScopeKey == scopeKey);
+        Assert.True(
+            match is not null,
+            $"expected a sample for ({metric}, {scope}, {scopeKey ?? "<null>"})");
+        Assert.Equal(value, match!.Value);
+    }
+
+    /// <summary>Captured arguments of one KPI computation call.</summary>
+    private readonly record struct KpiCall(DateTime StuckCutoff, DateTime IntervalSince);
+
+    /// <summary>
+    /// Hand-rolled <see cref="ISiteCallAuditRepository"/> stub returning
+    /// configurable snapshots and recording the cutoffs each KPI method received.
+    /// Non-KPI members are inert (this source only reads KPIs).
+    /// </summary>
+    private sealed class StubRepo : ISiteCallAuditRepository
+    {
+        public SiteCallKpiSnapshot Global { get; set; } =
+            new(0, 0, 0, 0, null, 0);
+        public IReadOnlyList<SiteCallSiteKpiSnapshot> PerSite { get; set; } =
+            Array.Empty<SiteCallSiteKpiSnapshot>();
+        public IReadOnlyList<SiteCallNodeKpiSnapshot> PerNode { get; set; } =
+            Array.Empty<SiteCallNodeKpiSnapshot>();
+
+        public List<KpiCall> Calls { get; } = new();
+
+        public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
+            DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default)
+        {
+            Calls.Add(new KpiCall(stuckCutoff, intervalSince));
+            return Task.FromResult(Global);
+        }
+
+        public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
+            DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default)
+        {
+            Calls.Add(new KpiCall(stuckCutoff, intervalSince));
+            return Task.FromResult(PerSite);
+        }
+
+        public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
+            DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default)
+        {
+            Calls.Add(new KpiCall(stuckCutoff, intervalSince));
+            return Task.FromResult(PerNode);
+        }
+
+        // ── Inert non-KPI members ──
+        public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) =>
+            Task.CompletedTask;
+
+        public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
+            Task.FromResult<SiteCall?>(null);
+
+        public Task<IReadOnlyList<SiteCall>> QueryAsync(
+            SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
+            Task.FromResult<IReadOnlyList<SiteCall>>(Array.Empty<SiteCall>());
+
+        public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
+            Task.FromResult(0);
+    }
+}