ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs

using System.Data.Common;
using Microsoft.EntityFrameworkCore.Diagnostics;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Notifications;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;

namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;

/// <summary>
/// Query-shape coverage for the Notification Outbox KPI snapshots (arch-review 04).
/// Each snapshot is at most three status-equality legs (live queue, parked, delivered
/// in the interval) — the shape before arch-review 04 issued 5-7 sequential COUNT round
/// trips and materialized the whole live queue to find the oldest row, and the single
/// OR'd aggregation that replaced it scanned the clustered index once the table held
/// ~1 M Parked rows (#122).
/// </summary>
public class NotificationOutboxRepositoryKpiQueryShapeTests
{
    private static Notification NewNotification(
        string sourceSiteId, string? sourceNode, NotificationStatus status,
        DateTimeOffset createdAt, DateTimeOffset? deliveredAt = null) =>
        new(Guid.NewGuid().ToString(), NotificationType.Email, "Ops List", "Subject", "Body", sourceSiteId)
        {
            Status = status,
            CreatedAt = createdAt,
            DeliveredAt = deliveredAt,
            SourceNode = sourceNode,
        };

    /// <summary>Verifies that all three KPI snapshots issue at most three seekable legs and never OR the parked and live predicates.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AllThreeKpiSnapshots_IssueAtMostThreeStatusLegs_NeverOneOrdScan()
    {
        var counter = new SelectCountingInterceptor();
        await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter);
        var now = DateTimeOffset.UtcNow;

        // Three non-terminal rows for one site/node — the "live queue" the old code
        // pulled into memory to find the oldest. All older than the stuck cutoff.
        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Pending, now.AddMinutes(-30)));
        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Retrying, now.AddMinutes(-20)));
        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Pending, now.AddMinutes(-10)));
        await ctx.SaveChangesAsync();

        var repo = new NotificationOutboxRepository(ctx);
        var stuckCutoff = now.AddMinutes(-5);
        var deliveredSince = now.AddMinutes(-60);

        // Global: three legs — live (counts + MIN), parked count, delivered count.
        counter.Reset();
        var global = await repo.ComputeKpisAsync(stuckCutoff, deliveredSince);
        Assert.True(counter.Count <= 3,
            $"ComputeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
        AssertNoLegMixesParkedWithLive(counter.Commands);
        // The aggregation must be PREDICATE-RESTRICTED, not an unrestricted
        // scan. Every KPI here concerns the live queue, the parked backlog or the
        // last delivery interval; historical Delivered/Discarded rows — which are
        // the overwhelming bulk of a retained table — contribute to none of them.
        // Without the WHERE the query's cost grows with retention rather than with
        // the working set. (An index-restricted aggregation was already the shape
        // of the per-site/per-node snapshots; the global one was the odd one out.)
        Assert.All(
            counter.Commands.Where(sql => sql.Contains("COUNT", StringComparison.OrdinalIgnoreCase)),
            sql => Assert.Contains("WHERE", sql, StringComparison.OrdinalIgnoreCase));
        Assert.Equal(3, global.QueueDepth);
        Assert.NotNull(global.OldestPendingAge);

        // Per-site: a single grouped conditional-aggregation query (counts + MIN).
        counter.Reset();
        var perSite = await repo.ComputePerSiteKpisAsync(stuckCutoff, deliveredSince);
        Assert.True(counter.Count <= 3,
            $"ComputePerSiteKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
        AssertNoLegMixesParkedWithLive(counter.Commands);
        var siteA = Assert.Single(perSite);
        Assert.Equal(3, siteA.QueueDepth);
        Assert.NotNull(siteA.OldestPendingAge);

        // Per-node: same single grouped aggregation, keyed by SourceNode.
        counter.Reset();
        var perNode = await repo.ComputePerNodeKpisAsync(stuckCutoff, deliveredSince);
        Assert.True(counter.Count <= 3,
            $"ComputePerNodeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
        AssertNoLegMixesParkedWithLive(counter.Commands);
        var nodeA = Assert.Single(perNode);
        Assert.Equal(3, nodeA.QueueDepth);
        Assert.NotNull(nodeA.OldestPendingAge);
    }

    /// <summary>Verifies the three legs merge per key and a key with only one leg still surfaces.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task PerSiteKpis_MergeLegsPerKey()
    {
        await using var ctx = SqliteTestHelper.CreateInMemoryContext(new SelectCountingInterceptor());
        var now = DateTimeOffset.UtcNow;
        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Pending, now.AddMinutes(-30)));
        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Parked, now.AddMinutes(-30)));
        ctx.Notifications.Add(NewNotification("plant-b", "node-b", NotificationStatus.Parked, now.AddMinutes(-30)));
        ctx.Notifications.Add(NewNotification("plant-c", "node-a", NotificationStatus.Delivered, now.AddMinutes(-30), now.AddMinutes(-1)));
        ctx.Notifications.Add(NewNotification("plant-d", "node-b", NotificationStatus.Discarded, now.AddMinutes(-30)));
        await ctx.SaveChangesAsync();

        var repo = new NotificationOutboxRepository(ctx);
        var sites = await repo.ComputePerSiteKpisAsync(now.AddMinutes(-5), now.AddMinutes(-60));

        Assert.Equal(new[] { "plant-a", "plant-b", "plant-c" }, sites.Select(s => s.SourceSiteId));
        Assert.Equal((1, 1, 1, 0), (sites[0].QueueDepth, sites[0].StuckCount, sites[0].ParkedCount, sites[0].DeliveredLastInterval));
        Assert.NotNull(sites[0].OldestPendingAge);
        Assert.Equal((0, 1, 0), (sites[1].QueueDepth, sites[1].ParkedCount, sites[1].DeliveredLastInterval));
        Assert.Null(sites[1].OldestPendingAge);
        Assert.Equal((0, 0, 1), (sites[2].QueueDepth, sites[2].ParkedCount, sites[2].DeliveredLastInterval));

        var global = await repo.ComputeKpisAsync(now.AddMinutes(-5), now.AddMinutes(-60));
        Assert.Equal((1, 1, 2, 1), (global.QueueDepth, global.StuckCount, global.ParkedCount, global.DeliveredLastInterval));
    }

    // An OR across the parked and live predicates is what made SQL Server scan the
    // clustered index on a large table (#122); each leg must name one status set.
    private static void AssertNoLegMixesParkedWithLive(IEnumerable<string> commands) =>
        Assert.All(commands, sql => Assert.False(
            sql.Contains("'Parked'", StringComparison.Ordinal) && sql.Contains("'Pending'", StringComparison.Ordinal),
            $"a KPI leg ORs Parked with the live queue:\n{sql}"));

    /// <summary>
    /// Records the text of every reader command touching the Notifications table so a
    /// test can count the round trips a KPI snapshot makes and inspect their shape.
    /// </summary>
    private sealed class SelectCountingInterceptor : DbCommandInterceptor
    {
        private readonly List<string> _commands = new();

        /// <summary>The text of every reader command recorded so far.</summary>
        public IReadOnlyList<string> Commands => _commands;

        /// <summary>The number of reader commands recorded so far.</summary>
        public int Count => _commands.Count;

        /// <summary>Clears all recorded commands.</summary>
        public void Reset() => _commands.Clear();

        private void Record(DbCommand command)
        {
            if (command.CommandText.Contains("Notifications", StringComparison.OrdinalIgnoreCase))
            {
                _commands.Add(command.CommandText);
            }
        }

        /// <inheritdoc />
        public override InterceptionResult<DbDataReader> ReaderExecuting(
            DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
        {
            Record(command);
            return base.ReaderExecuting(command, eventData, result);
        }

        /// <inheritdoc />
        public override ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(
            DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result,
            CancellationToken cancellationToken = default)
        {
            Record(command);
            return base.ReaderExecutingAsync(command, eventData, result, cancellationToken);
        }
    }
}