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

using System.Text.Json;
using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
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 Parquet archive (ScadaBridge#42) against the
/// real migrated, partitioned schema: archive-then-purge leaves the month's hash chain
/// verifiable and the archive re-verifiable against the purge anchor; the purge refuses to
/// drop a partition that changed after it was archived; tampering with an archive is caught.
/// </summary>
/// <remarks>
/// Every test owns a distinct month (partition) of the shared fixture database; the chain
/// and the per-channel purge are table-wide, so each test drains the chain itself.
/// </remarks>
public sealed class AuditLogPartitionArchiveTests : IClassFixture<MsSqlMigrationFixture>, IDisposable
{
    private readonly MsSqlMigrationFixture _fixture;
    private readonly DirectoryInfo _root = Directory.CreateTempSubdirectory("auditlog-archive-");

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

    /// <inheritdoc />
    public void Dispose() => _root.Delete(recursive: true);

    /// <summary>
    /// Archive, then purge: the archive holds exactly the partition's rows (chained,
    /// unchained, with their chain columns) and its tombstones; after the switch the month's
    /// chain still verifies from the anchor, and the archive verifies against that anchor.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task ArchiveThenPurge_KeepsTheChainVerifiable_AndTheArchiveMatchesTheAnchor()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 2);
        await InsertAsync(Enumerable.Range(1, 10).Select(i => NewEvent(month.AddDays(i).AddTicks(i * 7_777))));
        // Two rows of a short-retention channel, purged per channel mid-chain → tombstones.
        await InsertAsync([NewEvent(month.AddHours(1), AuditChannel.DbOutbound), NewEvent(month.AddHours(2), AuditChannel.DbOutbound)]);
        await DrainAsync();
        await using (var ctx = CreateContext())
        {
            Assert.Equal(2, await new AuditLogRepository(ctx).PurgeChannelOlderThanAsync("DbOutbound", month.AddDays(1), 100));
        }

        await InsertAsync([NewEvent(month.AddDays(20))]); // written after the drain: unchained
        var before = await PartitionRowsAsync(month);
        Assert.Equal(11, before.Count);

        var manifest = await ArchiveAsync(month);
        Assert.Equal(AuditArchiveStatus.Archived, manifest.Status);
        Assert.Equal(11, manifest.RowCount);
        Assert.Equal(10, manifest.ChainedRowCount);
        Assert.Equal(1, manifest.UnchainedRowCount);
        Assert.Equal(2, manifest.TombstoneCount);
        var chain = Assert.Single(manifest.Chains);
        Assert.Equal("2026-02", chain.Partition);
        Assert.Equal(12, chain.LastSeq);

        // Row equality: the archive's data file holds exactly the partition's stored rows.
        var archived = await ReadArchiveRowsAsync(manifest);
        Assert.Equal(before.Select(Key).Order(), archived.Select(Key).Order());

        await using (var ctx = CreateContext())
        {
            Assert.Equal(11, await Archive(ctx).PurgeArchivedPartitionAsync(manifest));
        }

        Assert.Empty(await PartitionRowsAsync(month));
        var monthChain = await VerifyMonthAsync(month);
        Assert.Equal(AuditChainStatus.Intact, monthChain.Status);
        Assert.Equal(12, monthChain.PurgedThroughSeq);

        var verification = await VerifyArchiveAsync(manifest.Name);
        Assert.True(verification.Verified, verification.FileProblem);
        Assert.Equal(AuditArchiveStatus.Purged, verification.ManifestStatus);
        Assert.Equal(11, verification.RowsRead);
        var v = Assert.Single(verification.Chains);
        Assert.Equal(AuditChainStatus.Intact, v.Chain.Status);
        Assert.Equal(10, v.Chain.RowsVerified);
        Assert.Equal(2, v.Chain.TombstonesApplied);
        Assert.Equal(1, v.Chain.UnchainedRows);
        Assert.Equal(AuditArchiveAnchorStatus.Matches, v.Anchor);

        await using (var ctx = CreateContext())
        {
            var listed = Assert.Single(await Archive(ctx).ListAsync(), m => m.Partition == "2026-02");
            Assert.Equal(AuditArchiveStatus.Purged, listed.Status);
            Assert.NotNull(listed.PurgedAtUtc);
        }
    }

    /// <summary>
    /// A row that lands in the partition after it was archived makes the purge refuse
    /// (nothing dropped); a fresh archive then purges. The stale archive still verifies as a
    /// complete, un-purged copy.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task RowArrivingAfterTheArchive_StopsThePurge()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 3);
        await InsertAsync(Enumerable.Range(1, 5).Select(i => NewEvent(month.AddDays(i))));
        await DrainAsync();
        var stale = await ArchiveAsync(month);

        await InsertAsync([NewEvent(month.AddDays(9))]);
        await using (var ctx = CreateContext())
        {
            var ex = await Assert.ThrowsAsync<AuditArchiveStaleException>(() => Archive(ctx).PurgeArchivedPartitionAsync(stale));
            Assert.Equal(AuditPartitionSwitchGuard.RowCountChangedError, ((SqlException)ex.InnerException!).Number);
        }

        Assert.Equal(6, (await PartitionRowsAsync(month)).Count);
        Assert.Null((await VerifyMonthAsync(month)).PurgedThroughSeq);

        var staleVerification = await VerifyArchiveAsync(stale.Name);
        Assert.True(staleVerification.Verified);
        Assert.Equal(AuditArchiveStatus.Archived, staleVerification.ManifestStatus);
        Assert.Equal(AuditArchiveAnchorStatus.NotPurged, Assert.Single(staleVerification.Chains).Anchor);

        // Positive control: a fresh archive holds the new row and is allowed to purge.
        var fresh = await ArchiveAsync(month);
        Assert.Equal(6, fresh.RowCount);
        await using (var ctx = CreateContext())
        {
            Assert.Equal(6, await Archive(ctx).PurgeArchivedPartitionAsync(fresh));
        }

        Assert.Empty(await PartitionRowsAsync(month));
    }

    /// <summary>
    /// A chaining pass between the archive and the purge (the row count is unchanged, the
    /// chain head moved) also stops the purge: otherwise the anchor would cover links the
    /// archive does not hold.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task ChainAdvancingAfterTheArchive_StopsThePurge()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 4);
        await InsertAsync(Enumerable.Range(1, 4).Select(i => NewEvent(month.AddDays(i))));
        await DrainAsync();
        await InsertAsync([NewEvent(month.AddDays(8))]); // unchained at archive time
        var manifest = await ArchiveAsync(month);
        Assert.Equal(1, manifest.UnchainedRowCount);

        await DrainAsync(); // links the row: head 4 → 5, row count unchanged
        await using (var ctx = CreateContext())
        {
            var ex = await Assert.ThrowsAsync<AuditArchiveStaleException>(() => Archive(ctx).PurgeArchivedPartitionAsync(manifest));
            Assert.Equal(AuditPartitionSwitchGuard.ChainAdvancedError, ((SqlException)ex.InnerException!).Number);
        }

        Assert.Equal(5, (await PartitionRowsAsync(month)).Count);
    }

    /// <summary>
    /// Tampering with a purged archive is caught three ways: a flipped byte (file hash), a
    /// re-written file with an edited row and a re-hashed manifest (chain), and a forger who
    /// also recomputes the whole chain (the purge anchor in the database no longer matches).
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task TamperedArchive_FailsVerification()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 5);
        await InsertAsync(Enumerable.Range(1, 8).Select(i => NewEvent(month.AddDays(i))));
        await DrainAsync();
        var manifest = await ArchiveAndPurgeAsync(month);
        Assert.True((await VerifyArchiveAsync(manifest.Name)).Verified);

        var dataPath = Path.Combine(_root.FullName, "2026", manifest.DataFile);
        var manifestPath = Path.Combine(_root.FullName, "2026", manifest.Name + ".manifest.json");
        var original = await File.ReadAllBytesAsync(dataPath);
        var originalManifest = await File.ReadAllTextAsync(manifestPath);

        // 1. A flipped byte.
        var flipped = (byte[])original.Clone();
        flipped[original.Length / 2] ^= 0xFF;
        await File.WriteAllBytesAsync(dataPath, flipped);
        var result = await VerifyArchiveAsync(manifest.Name);
        Assert.False(result.Verified);
        Assert.False(result.FilesIntact);
        Assert.Contains("SHA-256", result.FileProblem);

        // 2. An edited row in a re-written file, manifest hash/size updated to match.
        var rows = await ReadArchiveRowsAsync(manifest, original);
        var victim = rows[3];
        rows[3] = victim with { Target = "forged" };
        await RewriteAsync(rows, manifest, dataPath, manifestPath, lastHash: null);
        result = await VerifyArchiveAsync(manifest.Name);
        Assert.False(result.Verified);
        Assert.True(result.FilesIntact);
        var brk = Assert.Single(result.Chains).Chain.FirstBreak!;
        Assert.Equal(AuditChainBreakKind.HashMismatch, brk.Kind);
        Assert.Equal(victim.EventId, brk.EventId);

        // 3. The forger recomputes every link and the manifest's chain end: the archive is
        //    self-consistent, but its end no longer matches the anchor the purge left.
        var previous = AuditChainHasher.GenesisHash.ToArray();
        for (var i = 0; i < rows.Count; i++)
        {
            var hash = AuditChainHasher.ComputeRowHash(previous, month, rows[i].ChainSeq!.Value, rows[i]);
            rows[i] = rows[i] with { RowHash = hash };
            previous = hash;
        }

        await RewriteAsync(rows, manifest, dataPath, manifestPath, lastHash: Convert.ToHexStringLower(previous));
        result = await VerifyArchiveAsync(manifest.Name);
        Assert.False(result.Verified);
        var c = Assert.Single(result.Chains);
        Assert.Equal(AuditChainStatus.Intact, c.Chain.Status);
        Assert.Equal(AuditArchiveAnchorStatus.Mismatch, c.Anchor);

        // Positive control: the untouched archive still verifies.
        await File.WriteAllBytesAsync(dataPath, original);
        await File.WriteAllTextAsync(manifestPath, originalManifest);
        Assert.True((await VerifyArchiveAsync(manifest.Name)).Verified);
    }

    /// <summary>
    /// Rows that arrive for a month after it was purged chain on from the anchor; archiving
    /// and purging them again produces a second archive that resumes from the first one's end.
    /// The first archive then reports its anchor as superseded, and both verify.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task LateRows_AreArchivedAgain_FromTheAnchor()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        var month = Month(2026, 6);
        await InsertAsync(Enumerable.Range(1, 3).Select(i => NewEvent(month.AddDays(i))));
        await DrainAsync();
        var first = await ArchiveAndPurgeAsync(month);

        await InsertAsync([NewEvent(month.AddDays(15)), NewEvent(month.AddDays(16))]);
        await DrainAsync();
        await Task.Delay(5); // distinct archive name (millisecond stamp)
        var second = await ArchiveAndPurgeAsync(month);
        var chain = Assert.Single(second.Chains);
        Assert.Equal(3, chain.PriorPurgedThroughSeq);
        Assert.Equal(5, chain.LastSeq);

        var secondResult = await VerifyArchiveAsync(second.Name);
        Assert.True(secondResult.Verified);
        Assert.Equal(AuditArchiveAnchorStatus.Matches, Assert.Single(secondResult.Chains).Anchor);
        Assert.Equal(2, Assert.Single(secondResult.Chains).Chain.RowsVerified);

        var firstResult = await VerifyArchiveAsync(first.Name);
        Assert.True(firstResult.Verified);
        Assert.Equal(AuditArchiveAnchorStatus.Superseded, Assert.Single(firstResult.Chains).Anchor);
    }

    /// <summary>A name that is not an archive name (e.g. a path) is rejected before any file access.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]
    public async Task VerifyRejectsPathLikeNames()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
        await using var ctx = CreateContext();
        await Assert.ThrowsAsync<ArgumentException>(() => Archive(ctx).VerifyAsync("../../etc/passwd"));
        await Assert.ThrowsAsync<FileNotFoundException>(() => Archive(ctx).VerifyAsync("auditlog-2026-01-20260101T000000000Z"));
    }

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

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

    private AuditLogPartitionArchive Archive(ScadaBridgeDbContext ctx) => new(ctx, new AuditLogArchiveOptions
    {
        Enabled = true,
        Directory = _root.FullName,
        RowGroupSize = 1_000,
    });

    private async Task<AuditArchiveManifest> ArchiveAsync(DateTime month)
    {
        await using var ctx = CreateContext();
        return await Archive(ctx).ArchivePartitionAsync(month);
    }

    private async Task<AuditArchiveManifest> ArchiveAndPurgeAsync(DateTime month)
    {
        var manifest = await ArchiveAsync(month);
        await using var ctx = CreateContext();
        await Archive(ctx).PurgeArchivedPartitionAsync(manifest);
        return manifest;
    }

    private async Task<AuditArchiveVerificationResult> VerifyArchiveAsync(string name)
    {
        await using var ctx = CreateContext();
        return await Archive(ctx).VerifyAsync(name);
    }

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

    private async Task<List<AuditChainRow>> ReadArchiveRowsAsync(AuditArchiveManifest manifest, byte[]? bytes = null)
    {
        bytes ??= await File.ReadAllBytesAsync(Path.Combine(_root.FullName, manifest.Partition[..4], manifest.DataFile));
        var rows = new List<AuditChainRow>();
        await foreach (var row in AuditArchiveParquet.ReadRowsAsync(new MemoryStream(bytes)))
        {
            rows.Add(row);
        }

        return rows;
    }

    private static async Task RewriteAsync(
        List<AuditChainRow> rows, AuditArchiveManifest manifest, string dataPath, string manifestPath, string? lastHash)
    {
        var stream = new MemoryStream();
        await using (var writer = await AuditArchiveParquet.CreateDataWriterAsync(stream, manifest.Partition, 1_000))
        {
            foreach (var row in rows)
            {
                await writer.AddAsync(row);
            }
        }

        var bytes = stream.ToArray();
        await File.WriteAllBytesAsync(dataPath, bytes);
        var forged = manifest with
        {
            Status = AuditArchiveStatus.Purged,
            DataSha256 = Convert.ToHexStringLower(System.Security.Cryptography.SHA256.HashData(bytes)),
            DataBytes = bytes.Length,
            Chains = lastHash is null ? manifest.Chains : [manifest.Chains[0] with { LastHash = lastHash }],
        };
        await File.WriteAllTextAsync(manifestPath, JsonSerializer.Serialize(forged, new JsonSerializerOptions
        {
            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
        }));
    }

    private async Task<List<AuditChainRow>> PartitionRowsAsync(DateTime month)
    {
        await using var conn = new SqlConnection(_fixture.ConnectionString);
        await conn.OpenAsync();
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = $"SELECT {AuditLogHashChain.RowColumns}, ChainSeq, RowHash FROM dbo.AuditLog "
            + "WHERE OccurredAtUtc >= @s AND OccurredAtUtc < @e;";
        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 rows = new List<AuditChainRow>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            rows.Add(AuditLogHashChain.ReadRow(reader, withLink: true));
        }

        return rows;
    }

    private static string Key(AuditChainRow r) => string.Join('|',
        r.EventId, r.OccurredAtUtc.Ticks, r.Actor ?? "<null>", r.Action, r.Outcome, r.Category, r.Target ?? "<null>",
        r.SourceNode ?? "<null>", r.CorrelationId?.ToString() ?? "<null>", r.DetailsJson ?? "<null>",
        r.ChainSeq?.ToString() ?? "<null>", r.RowHash is null ? "<null>" : Convert.ToHexString(r.RowHash));

    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);
            if (r.RowsChained == 0 && !r.LockBusy)
            {
                return;
            }
        }

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

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

    private static AuditEvent NewEvent(DateTime occurredAtUtc, AuditChannel channel = AuditChannel.ApiOutbound) =>
        ScadaBridgeAuditEventFactory.Create(
            channel: channel,
            kind: channel == AuditChannel.DbOutbound ? AuditKind.DbWrite : AuditKind.ApiCall,
            status: AuditStatus.Delivered,
            occurredAtUtc: occurredAtUtc,
            actor: "archive-test",
            target: "erp",
            sourceNode: "central-a",
            sourceSiteId: "archive-test",
            correlationId: Guid.NewGuid());
}