ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Migrations/FlipAuditLogClusteredKeyMigrationTests.cs

using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
using Microsoft.EntityFrameworkCore.Migrations.Operations;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using Xunit.Abstractions;

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

/// <summary>
/// SQL Server tests for the <c>FlipAuditLogClusteredKey</c> migration (ScadaBridge#111,
/// variant B2 of <c>docs/plans/2026-11-09-auditlog-clustered-key-benchmark.md</c>): the new
/// clustered key and the aligned unique EventId index, every other index and grant preserved,
/// a faithful Down, a re-runnable Up, a partition-switch purge that works on BOTH key shapes,
/// and the read win itself — the Audit Log page's page-1 and keyset queries read a small
/// fraction of the pages they read on the old key.
/// </summary>
/// <remarks>
/// The tests in this class migrate the fixture database down and back up, so they own their
/// fixture (one database per test class) and every test leaves the database fully migrated.
/// </remarks>
public sealed class FlipAuditLogClusteredKeyMigrationTests : IClassFixture<MsSqlMigrationFixture>
{
    private const string FlipMigration = "20261110033732_FlipAuditLogClusteredKey";

    // The migration immediately before the flip, so "Down" reverts exactly this one.
    private static readonly string PreviousMigration = MigrationBefore(FlipMigration);

    private static readonly string[] NonClusteredIndexes =
    [
        "IX_AuditLog_OccurredAtUtc", "IX_AuditLog_Site_Occurred", "IX_AuditLog_CorrelationId",
        "IX_AuditLog_Channel_Status_Occurred", "IX_AuditLog_Target_Occurred", "IX_AuditLog_Execution",
        "IX_AuditLog_ParentExecution", "IX_AuditLog_Node_Occurred", "IX_AuditLog_Kind_Occurred",
        "IX_AuditLog_Actor_Occurred", "IX_AuditLog_Unchained", "IX_AuditLog_ChainSeq",
    ];

    private readonly MsSqlMigrationFixture _fixture;
    private readonly ITestOutputHelper _output;

    /// <summary>Creates the test class with its own migrated SQL Server database.</summary>
    /// <param name="fixture">The per-class migration fixture.</param>
    /// <param name="output">Test output (records the measured logical reads).</param>
    public FlipAuditLogClusteredKeyMigrationTests(MsSqlMigrationFixture fixture, ITestOutputHelper output)
    {
        _fixture = fixture;
        _output = output;
    }

    /// <summary>
    /// After the full chain: clustered PK (OccurredAtUtc, EventId), the unique aligned
    /// (EventId, OccurredAtUtc) index, all twelve non-clustered indexes with unchanged keys and
    /// filters, everything on ps_AuditLog_Month, and every append-only/chain grant intact.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task Up_FlipsTheClusteredKey_AddsTheAlignedUniqueIndex_AndKeepsEveryIndexAndGrant()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        var indexes = await ReadIndexesAsync();
        Assert.Equal("OccurredAtUtc, EventId", indexes["PK_AuditLog"].Keys);
        Assert.True(indexes["PK_AuditLog"].Clustered);
        Assert.Equal("EventId, OccurredAtUtc", indexes["UX_AuditLog_EventId_Occurred"].Keys);
        Assert.True(indexes["UX_AuditLog_EventId_Occurred"].Unique);
        Assert.Equal(NonClusteredIndexes.Length + 2, indexes.Count);
        Assert.All(indexes.Values, i => Assert.Equal("ps_AuditLog_Month", i.DataSpace));
        AssertTwelveNonClusteredIndexes(indexes);
        Assert.Equal(ExpectedGrants, await ReadGrantsAsync());
    }

    /// <summary>
    /// Down restores the old key and drops the B2 index; Up re-flips; re-running Up's SQL on an
    /// already-flipped table is a no-op; and a partition-switch purge succeeds on each shape
    /// (the staging table's key follows the live table's).
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task DownUp_RoundTrips_IsReRunnable_AndThePartitionSwitchPurgesEitherShape()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        await MigrateToAsync(PreviousMigration);
        var down = await ReadIndexesAsync();
        Assert.Equal("EventId, OccurredAtUtc", down["PK_AuditLog"].Keys);
        Assert.False(down.ContainsKey("UX_AuditLog_EventId_Occurred"));
        AssertTwelveNonClusteredIndexes(down);
        Assert.Equal(GrantsBeforeMaintenanceGrants, await ReadGrantsAsync());

        // Old shape: the purge's staging table must take the old key order.
        await AssertSwitchOutPurgesAsync(new DateTime(2027, 3, 1, 0, 0, 0, DateTimeKind.Utc));

        // Back to head (not just to the flip): migrations after it — AuditMaintenanceGrants —
        // were reverted by the Down too, and the class leaves the database fully migrated.
        await MigrateToHeadAsync();
        var up = await ReadIndexesAsync();
        Assert.Equal("OccurredAtUtc, EventId", up["PK_AuditLog"].Keys);

        // Re-run Up's SQL against the flipped table: guarded, so nothing changes.
        await using (var conn = _fixture.OpenConnection())
        {
            foreach (var op in new FlipAuditLogClusteredKey().UpOperations.OfType<SqlOperation>())
            {
                await using var cmd = new SqlCommand(op.Sql, conn) { CommandTimeout = 300 };
                await cmd.ExecuteNonQueryAsync();
            }
        }

        Assert.Equal(up.Select(kv => (kv.Key, kv.Value)).OrderBy(x => x.Key),
            (await ReadIndexesAsync()).Select(kv => (kv.Key, kv.Value)).OrderBy(x => x.Key));

        // New shape: same purge, now with the time-leading staging key.
        await AssertSwitchOutPurgesAsync(new DateTime(2027, 4, 1, 0, 0, 0, DateTimeKind.Utc));
        Assert.Equal(ExpectedGrants, await ReadGrantsAsync());
    }

    /// <summary>
    /// The read win, measured with the session's logical-read counter around the repository's real query: page 1
    /// (newest 50, no filter) and a keyset page read far fewer AuditLog pages on the flipped
    /// key than on the old one over the same rows.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task PageOneAndKeysetPage_ReadAFractionOfThePages_OnTheFlippedKey()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        await SeedRowsAsync(new DateTime(2027, 6, 1, 0, 0, 0, DateTimeKind.Utc), rows: 20_000);

        var (flippedPage1, flippedKeyset) = await MeasurePageReadsAsync();
        await MigrateToAsync(PreviousMigration);
        try
        {
            var (oldPage1, oldKeyset) = await MeasurePageReadsAsync();
            _output.WriteLine(
                $"logical reads, 20k rows - page 1: {flippedPage1} flipped vs {oldPage1} old; keyset: {flippedKeyset} flipped vs {oldKeyset} old");
            Assert.True(flippedPage1 * 10 < oldPage1,
                $"page 1: {flippedPage1} logical reads on (OccurredAtUtc, EventId) vs {oldPage1} on (EventId, OccurredAtUtc)");
            Assert.True(flippedKeyset * 10 < oldKeyset,
                $"keyset: {flippedKeyset} logical reads on (OccurredAtUtc, EventId) vs {oldKeyset} on (EventId, OccurredAtUtc)");
        }
        finally
        {
            await MigrateToHeadAsync();
        }
    }

    private static readonly string[] ExpectedGrants =
    [
        "scadabridge_audit_chainer|SELECT|GRANT|",
        "scadabridge_audit_chainer|UPDATE|GRANT|ChainSeq",
        "scadabridge_audit_chainer|UPDATE|GRANT|RowHash",
        "scadabridge_audit_purger|DELETE|GRANT|",
        "scadabridge_audit_purger|SELECT|GRANT|",
        // AuditMaintenanceGrants (migration 64, after the flip) — the archive path's column grant.
        "scadabridge_audit_purger|UPDATE|GRANT|SourceNode",
        "scadabridge_audit_writer|DELETE|DENY|",
        "scadabridge_audit_writer|INSERT|GRANT|",
        "scadabridge_audit_writer|SELECT|GRANT|",
        "scadabridge_audit_writer|UPDATE|DENY|",
    ];

    // The grants one migration before the flip: AuditMaintenanceGrants is not applied yet.
    private static readonly string[] GrantsBeforeMaintenanceGrants =
        ExpectedGrants.Where(g => !g.EndsWith("|SourceNode", StringComparison.Ordinal)).ToArray();

    private static void AssertTwelveNonClusteredIndexes(IReadOnlyDictionary<string, IndexShape> indexes)
    {
        Assert.Equal("OccurredAtUtc DESC", indexes["IX_AuditLog_OccurredAtUtc"].Keys);
        Assert.Equal("Status", indexes["IX_AuditLog_OccurredAtUtc"].Included);
        Assert.Equal("SourceSiteId, OccurredAtUtc DESC", indexes["IX_AuditLog_Site_Occurred"].Keys);
        Assert.Equal("CorrelationId", indexes["IX_AuditLog_CorrelationId"].Keys);
        Assert.Equal("([CorrelationId] IS NOT NULL)", indexes["IX_AuditLog_CorrelationId"].Filter);
        Assert.Equal("Category, Status, OccurredAtUtc DESC", indexes["IX_AuditLog_Channel_Status_Occurred"].Keys);
        Assert.Equal("Target, OccurredAtUtc DESC", indexes["IX_AuditLog_Target_Occurred"].Keys);
        Assert.Equal("([Target] IS NOT NULL)", indexes["IX_AuditLog_Target_Occurred"].Filter);
        Assert.Equal("ExecutionId", indexes["IX_AuditLog_Execution"].Keys);
        Assert.Equal("ParentExecutionId", indexes["IX_AuditLog_ParentExecution"].Keys);
        Assert.Equal("SourceNode, OccurredAtUtc", indexes["IX_AuditLog_Node_Occurred"].Keys);
        Assert.Equal("Kind, OccurredAtUtc DESC", indexes["IX_AuditLog_Kind_Occurred"].Keys);
        Assert.Equal("Actor, OccurredAtUtc DESC", indexes["IX_AuditLog_Actor_Occurred"].Keys);
        Assert.Equal("OccurredAtUtc, EventId", indexes["IX_AuditLog_Unchained"].Keys);
        Assert.Equal("([ChainSeq] IS NULL)", indexes["IX_AuditLog_Unchained"].Filter);
        Assert.Equal("ChainSeq", indexes["IX_AuditLog_ChainSeq"].Keys);
        Assert.Equal("([ChainSeq] IS NOT NULL)", indexes["IX_AuditLog_ChainSeq"].Filter);
        Assert.All(NonClusteredIndexes, n => Assert.False(indexes[n].Clustered, n));
    }

    private async Task AssertSwitchOutPurgesAsync(DateTime month)
    {
        await SeedRowsAsync(month, rows: 50);
        await using var ctx = CreateContext();
        var purged = await new AuditLogRepository(ctx).SwitchOutPartitionAsync(month);
        Assert.Equal(50, purged);
        await using var conn = _fixture.OpenConnection();
        await using var cmd = new SqlCommand(
            "SELECT COUNT_BIG(*) FROM dbo.AuditLog WHERE OccurredAtUtc >= @s AND OccurredAtUtc < @e;", conn);
        cmd.Parameters.AddWithValue("@s", month);
        cmd.Parameters.AddWithValue("@e", month.AddMonths(1));
        Assert.Equal(0L, (long)(await cmd.ExecuteScalarAsync())!);
    }

    private async Task<(long Page1, long Keyset)> MeasurePageReadsAsync()
    {
        await using var conn = new SqlConnection(_fixture.ConnectionString);
        await conn.OpenAsync();

        // EF reuses the already-open connection, so the session's cumulative logical-read
        // counter (sys.dm_exec_sessions) brackets exactly the repository's query. The probe
        // itself reads no AuditLog pages.
        await using var ctx = new ScadaBridgeDbContext(
            new DbContextOptionsBuilder<ScadaBridgeDbContext>().UseSqlServer(conn).Options);
        var repo = new AuditLogRepository(ctx);

        var before = await SessionLogicalReadsAsync(conn);
        var page1 = await repo.QueryAsync(new AuditLogQueryFilter(), new AuditLogPaging(50));
        var page1Reads = await SessionLogicalReadsAsync(conn) - before;
        Assert.Equal(50, page1.Count);

        var cursor = page1[^1];
        before = await SessionLogicalReadsAsync(conn);
        var page2 = await repo.QueryAsync(
            new AuditLogQueryFilter(),
            new AuditLogPaging(50, cursor.OccurredAtUtc.UtcDateTime, cursor.EventId));
        var keysetReads = await SessionLogicalReadsAsync(conn) - before;
        Assert.Equal(50, page2.Count);
        Assert.True(page2[0].OccurredAtUtc <= cursor.OccurredAtUtc);
        return (page1Reads, keysetReads);
    }

    private static async Task<long> SessionLogicalReadsAsync(SqlConnection conn)
    {
        await using var cmd = new SqlCommand("SELECT logical_reads FROM sys.dm_exec_sessions WHERE session_id = @@SPID;", conn);
        return Convert.ToInt64(await cmd.ExecuteScalarAsync());
    }

    private async Task SeedRowsAsync(DateTime month, int rows)
    {
        // Set-based seed with random EventIds (as minted at the sites) spread over the month;
        // DetailsJson carries kind/status so the persisted computed columns materialise.
        await using var conn = _fixture.OpenConnection();
        await using var cmd = new SqlCommand(@"
WITH n AS (SELECT TOP (@rows) ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS i
           FROM sys.all_objects a CROSS JOIN sys.all_objects b)
INSERT INTO dbo.AuditLog (EventId, OccurredAtUtc, Actor, Action, Outcome, Category, Target, SourceNode, CorrelationId, DetailsJson)
SELECT NEWID(), DATEADD(SECOND, CAST(i * (2419200.0 / @rows) AS int), @month), N'flip-test', 'ApiOutbound.ApiCall', 'Success',
       'ApiOutbound', N'erp', 'central-a', NULL,
       N'{""kind"":""ApiCall"",""status"":""Delivered"",""sourceSiteId"":""flip"",""pad"":""' + REPLICATE(N'x', 400) + N'""}'
FROM n;", conn) { CommandTimeout = 300 };
        cmd.Parameters.AddWithValue("@rows", rows);
        cmd.Parameters.Add(new SqlParameter("@month", System.Data.SqlDbType.DateTime2) { Scale = 7, Value = month });
        await cmd.ExecuteNonQueryAsync();
    }

    private async Task MigrateToAsync(string migration)
    {
        await using var ctx = new ScadaBridgeDbContext(new DbContextOptionsBuilder<ScadaBridgeDbContext>()
            .UseSqlServer(_fixture.ConnectionString, sql => sql.CommandTimeout(600)).Options);
        await ctx.GetService<IMigrator>().MigrateAsync(migration);
    }

    private async Task MigrateToHeadAsync()
    {
        await using var ctx = new ScadaBridgeDbContext(new DbContextOptionsBuilder<ScadaBridgeDbContext>()
            .UseSqlServer(_fixture.ConnectionString, sql => sql.CommandTimeout(600)).Options);
        await ctx.GetService<IMigrator>().MigrateAsync();
    }

    private static string MigrationBefore(string migration)
    {
        using var ctx = new ScadaBridgeDbContext(new DbContextOptionsBuilder<ScadaBridgeDbContext>()
            .UseSqlServer("Server=unused;Database=unused").Options);
        var all = ctx.Database.GetMigrations().ToList();
        return all[all.IndexOf(migration) - 1];
    }

    private ScadaBridgeDbContext CreateContext()
        => new(new DbContextOptionsBuilder<ScadaBridgeDbContext>().UseSqlServer(_fixture.ConnectionString).Options);

    private async Task<Dictionary<string, IndexShape>> ReadIndexesAsync()
    {
        await using var conn = _fixture.OpenConnection();
        await using var cmd = new SqlCommand(@"
SELECT i.name, i.type, i.is_unique, ds.name, i.filter_definition,
  STUFF((SELECT ', ' + c.name + CASE WHEN ic.is_descending_key = 1 THEN ' DESC' ELSE '' END
         FROM sys.index_columns ic JOIN sys.columns c ON c.object_id = ic.object_id AND c.column_id = ic.column_id
         WHERE ic.object_id = i.object_id AND ic.index_id = i.index_id AND ic.key_ordinal > 0
         ORDER BY ic.key_ordinal FOR XML PATH('')), 1, 2, ''),
  STUFF((SELECT ', ' + c.name
         FROM sys.index_columns ic JOIN sys.columns c ON c.object_id = ic.object_id AND c.column_id = ic.column_id
         WHERE ic.object_id = i.object_id AND ic.index_id = i.index_id AND ic.is_included_column = 1
         ORDER BY ic.index_column_id FOR XML PATH('')), 1, 2, '')
FROM sys.indexes i JOIN sys.data_spaces ds ON ds.data_space_id = i.data_space_id
WHERE i.object_id = OBJECT_ID('dbo.AuditLog') AND i.type IN (1, 2);", conn);
        var result = new Dictionary<string, IndexShape>(StringComparer.Ordinal);
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            result[reader.GetString(0)] = new IndexShape(
                Clustered: reader.GetByte(1) == 1,
                Unique: reader.GetBoolean(2),
                DataSpace: reader.GetString(3),
                Filter: reader.IsDBNull(4) ? null : reader.GetString(4),
                Keys: reader.GetString(5),
                Included: reader.IsDBNull(6) ? null : reader.GetString(6));
        }

        return result;
    }

    private async Task<string[]> ReadGrantsAsync()
    {
        await using var conn = _fixture.OpenConnection();
        await using var cmd = new SqlCommand(@"
SELECT pr.name, p.permission_name, p.state_desc, COL_NAME(p.major_id, p.minor_id)
FROM sys.database_permissions p JOIN sys.database_principals pr ON pr.principal_id = p.grantee_principal_id
WHERE p.major_id = OBJECT_ID('dbo.AuditLog') AND p.class = 1;", conn);
        var list = new List<string>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            list.Add($"{reader.GetString(0)}|{reader.GetString(1)}|{reader.GetString(2)}|{(reader.IsDBNull(3) ? "" : reader.GetString(3))}");
        }

        return list.Order(StringComparer.Ordinal).ToArray();
    }

    private sealed record IndexShape(bool Clustered, bool Unique, string DataSpace, string? Filter, string Keys, string? Included);
}