ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Maintenance/AuditLogHashChainTests.cs

using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Migrations;

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

/// <summary>
/// SQL Server integration tests for the AuditLog hash chain (ScadaBridge#41) against the
/// real migrated, partitioned schema: continuity under concurrent writers and chainers,
/// detection of an edited / deleted / reordered row, and survival of both purge paths.
/// </summary>
/// <remarks>
/// The fixture database is shared by the class and the chain is per month across the whole
/// table, so every test owns a distinct month (xunit runs one class's tests sequentially).
/// Each test drains the chain before tampering so no other month is left unchained.
/// </remarks>
public class AuditLogHashChainTests : IClassFixture<MsSqlMigrationFixture>
{
    private readonly MsSqlMigrationFixture _fixture;

    /// <summary>Creates the test class with the shared migrated SQL Server fixture.</summary>
    /// <param name="fixture">The shared migration fixture.</param>
    public AuditLogHashChainTests(MsSqlMigrationFixture fixture) => _fixture = fixture;

    /// <summary>
    /// Many concurrent INSERTs (the two central nodes + site ingest) racing two concurrent
    /// chainers (a dual-active central pair) still produce one gap-free, intact chain.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task ConcurrentWritersAndChainers_ProduceOneIntactGapFreeChain()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 2);
        const int writers = 8;
        const int rowsPerWriter = 40;

        using var stop = new CancellationTokenSource();
        var lockBusySeen = 0;
        async Task ChainLoop()
        {
            while (!stop.IsCancellationRequested)
            {
                await using var ctx = CreateContext();
                var r = await new AuditLogHashChain(ctx).ChainPendingAsync(17, chainBacklog: true);
                if (r.LockBusy) Interlocked.Increment(ref lockBusySeen);
                if (r.RowsChained == 0) await Task.Delay(5);
            }
        }

        var chainers = new[] { Task.Run(ChainLoop), Task.Run(ChainLoop) };
        var writerTasks = Enumerable.Range(0, writers).Select(w => Task.Run(async () =>
        {
            for (var i = 0; i < rowsPerWriter; i++)
            {
                await using var ctx = CreateContext();
                // Out-of-order OccurredAtUtc on purpose: late-arriving site rows.
                var at = month.AddDays((w * 7 + i * 13) % 27).AddSeconds(i);
                await new AuditLogRepository(ctx).InsertIfNotExistsAsync(NewEvent(at, sourceNode: $"node-{w}"));
            }
        })).ToArray();

        await Task.WhenAll(writerTasks);
        await Task.Delay(200);
        await stop.CancelAsync();
        await Task.WhenAll(chainers);
        await DrainAsync();

        var result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Intact, result.Status);
        Assert.Equal(writers * rowsPerWriter, result.RowsVerified);
        Assert.Equal(writers * rowsPerWriter, result.HeadSeq);
        Assert.Equal(0, result.UnchainedRows);

        var seqs = await ChainSeqsAsync(month);
        Assert.Equal(Enumerable.Range(1, writers * rowsPerWriter).Select(i => (long)i), seqs);
    }

    /// <summary>Editing a chained row's content is reported at that row.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task EditedRow_IsDetected_WithRowIdAndPartition()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 3);
        var events = await SeedAndChainAsync(month, 6);
        Assert.Equal(AuditChainStatus.Intact, (await VerifyAsync(month)).Status);

        var victim = events[3];
        await ExecAsync($"UPDATE dbo.AuditLog SET Target = N'forged' WHERE EventId = '{victim}';");

        var result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Broken, result.Status);
        Assert.Equal(AuditChainBreakKind.HashMismatch, result.FirstBreak!.Kind);
        Assert.Equal("2026-03", result.FirstBreak.Partition);
        Assert.Equal(victim, result.FirstBreak.EventId);
        Assert.Equal(await SeqOfAsync(victim), result.FirstBreak.ChainSeq);
    }

    /// <summary>Deleting a chained row outside the purge path is reported as a missing sequence number.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task DeletedRow_IsDetected()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 4);
        var events = await SeedAndChainAsync(month, 6);
        var victim = events[2];
        var victimSeq = await SeqOfAsync(victim);

        await ExecAsync($"DELETE FROM dbo.AuditLog WHERE EventId = '{victim}';");

        var result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Broken, result.Status);
        Assert.Equal(AuditChainBreakKind.MissingRow, result.FirstBreak!.Kind);
        Assert.Equal(victimSeq, result.FirstBreak.ChainSeq);
        Assert.Equal("2026-04", result.FirstBreak.Partition);
    }

    /// <summary>Swapping two rows' chain positions is reported.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task ReorderedRows_AreDetected()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 5);
        var events = await SeedAndChainAsync(month, 6);
        var a = events[1];
        var b = events[4];
        var seqA = await SeqOfAsync(a);
        var seqB = await SeqOfAsync(b);

        await ExecAsync(
            $"UPDATE dbo.AuditLog SET ChainSeq = CASE WHEN EventId = '{a}' THEN {seqB} ELSE {seqA} END " +
            $"WHERE EventId IN ('{a}', '{b}');");

        var result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Broken, result.Status);
        Assert.Equal(AuditChainBreakKind.HashMismatch, result.FirstBreak!.Kind);
        Assert.Equal(Math.Min(seqA, seqB), result.FirstBreak.ChainSeq);
    }

    /// <summary>
    /// The per-channel retention DELETE removes chained rows from the middle of a month and
    /// the chain still verifies (tombstones); a later chaining pass continues the chain.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task PerChannelPurge_LeavesAVerifiableChain()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 6);
        var rows = new List<AuditEvent>();
        for (var i = 0; i < 10; i++)
        {
            rows.Add(NewEvent(month.AddDays(i + 1),
                channel: i % 2 == 0 ? AuditChannel.DbOutbound : AuditChannel.ApiOutbound,
                kind: i % 2 == 0 ? AuditKind.DbWrite : AuditKind.ApiCall));
        }

        await InsertAsync(rows);
        await DrainAsync();

        await using (var ctx = CreateContext())
        {
            var purged = await new AuditLogRepository(ctx).PurgeChannelOlderThanAsync(
                nameof(AuditChannel.DbOutbound), month.AddDays(8), batchSize: 2);
            Assert.Equal(4, purged); // days 1,3,5,7
        }

        var result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Intact, result.Status);
        Assert.Equal(4, result.TombstonesApplied);
        Assert.Equal(6, result.RowsVerified);

        // A row arriving after the purge chains on from the existing head.
        await InsertAsync([NewEvent(month.AddDays(20))]);
        await DrainAsync();
        result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Intact, result.Status);
        Assert.Equal(11, result.HeadSeq);
    }

    /// <summary>
    /// The monthly partition-switch purge removes the whole month; the head's checkpoint
    /// anchor keeps the month verifiable, including for a late row chained afterwards.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task PartitionSwitchPurge_LeavesAnAnchoredVerifiableChain()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 7);
        await SeedAndChainAsync(month, 5);

        await using (var ctx = CreateContext())
        {
            await new AuditLogRepository(ctx).SwitchOutPartitionAsync(month);
        }

        var result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Intact, result.Status);
        Assert.Equal(5, result.PurgedThroughSeq);
        Assert.Equal(0, result.RowsVerified);

        await InsertAsync([NewEvent(month.AddDays(3))]);
        await DrainAsync();
        result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Intact, result.Status);
        Assert.Equal(1, result.RowsVerified);
        Assert.Equal(6, result.HeadSeq);
    }

    /// <summary>
    /// The SourceNode backfill (the other sanctioned AuditLog UPDATE) must not touch a chained
    /// row — SourceNode is inside its hash.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task SourceNodeBackfill_SkipsChainedRows()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 8);
        await InsertAsync([NewEvent(month.AddDays(2), sourceNode: null)]);
        await DrainAsync();
        var unchained = NewEvent(month.AddDays(3), sourceNode: null);

        // A row written after the drain stays unchained until the next pass.
        await InsertAsync([unchained]);
        await using (var ctx = CreateContext())
        {
            await new AuditLogRepository(ctx).BackfillSourceNodeAsync("unknown", month.AddMonths(1), batchSize: 10);
        }

        await DrainAsync();
        var result = await VerifyAsync(month);
        Assert.Equal(AuditChainStatus.Intact, result.Status);
        Assert.Equal(2, result.RowsVerified);
        Assert.Equal("unknown", await ScalarAsync<string>($"SELECT SourceNode FROM dbo.AuditLog WHERE EventId = '{unchained.EventId}';"));
    }

    // ── helpers ──────────────────────────────────────────────────────────────

    private static DateTime Month(int year, int month) => new(year, month, 1, 0, 0, 0, DateTimeKind.Utc);

    private async Task<List<Guid>> SeedAndChainAsync(DateTime month, int count)
    {
        var events = Enumerable.Range(1, count).Select(i => NewEvent(month.AddDays(i).AddMinutes(i))).ToList();
        await InsertAsync(events);
        await DrainAsync();
        return events.Select(e => e.EventId).ToList();
    }

    private async Task InsertAsync(IEnumerable<AuditEvent> events)
    {
        await using var ctx = CreateContext();
        var repo = new AuditLogRepository(ctx);
        foreach (var e in events)
        {
            await repo.InsertIfNotExistsAsync(e);
        }
    }

    private async Task DrainAsync()
    {
        await using var ctx = CreateContext();
        var chain = new AuditLogHashChain(ctx);
        for (var i = 0; i < 1000; i++)
        {
            var r = await chain.ChainPendingAsync(500, chainBacklog: true);
            if (r.RowsChained == 0 && !r.LockBusy)
            {
                return;
            }
        }

        throw new InvalidOperationException("Chain did not drain.");
    }

    private async Task<AuditChainVerificationResult> VerifyAsync(DateTime month)
    {
        await using var ctx = CreateContext();
        return await new AuditLogHashChain(ctx).VerifyMonthAsync(month);
    }

    private async Task<long> SeqOfAsync(Guid eventId)
        => await ScalarAsync<long>($"SELECT ChainSeq FROM dbo.AuditLog WHERE EventId = '{eventId}';");

    private async Task<List<long>> ChainSeqsAsync(DateTime month)
    {
        await using var conn = new SqlConnection(_fixture.ConnectionString);
        await conn.OpenAsync();
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = "SELECT ChainSeq FROM dbo.AuditLog WHERE OccurredAtUtc >= @s AND OccurredAtUtc < @e ORDER BY ChainSeq;";
        cmd.Parameters.Add(new SqlParameter("@s", System.Data.SqlDbType.DateTime2) { Value = month });
        cmd.Parameters.Add(new SqlParameter("@e", System.Data.SqlDbType.DateTime2) { Value = month.AddMonths(1) });
        var list = new List<long>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            list.Add(reader.GetInt64(0));
        }

        return list;
    }

    private async Task ExecAsync(string sql)
    {
        await using var conn = new SqlConnection(_fixture.ConnectionString);
        await conn.OpenAsync();
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = sql;
        await cmd.ExecuteNonQueryAsync();
    }

    private async Task<T> ScalarAsync<T>(string sql)
    {
        await using var conn = new SqlConnection(_fixture.ConnectionString);
        await conn.OpenAsync();
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = sql;
        return (T)(await cmd.ExecuteScalarAsync())!;
    }

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

    private static AuditEvent NewEvent(
        DateTime occurredAtUtc,
        AuditChannel channel = AuditChannel.ApiOutbound,
        AuditKind kind = AuditKind.ApiCall,
        string? sourceNode = "central-a") =>
        ScadaBridgeAuditEventFactory.Create(
            channel: channel,
            kind: kind,
            status: AuditStatus.Delivered,
            occurredAtUtc: occurredAtUtc,
            target: "erp",
            sourceNode: sourceNode,
            sourceSiteId: "hash-chain-test");
}