lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/RetentionTests.cs

using Microsoft.Data.SqlClient;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;

[Collection(SqlHistorianCollection.Name)]
public sealed class RetentionTests(SqlHistorianFixture fx)
{
    private static readonly DateTime Now = new(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc);

    private sealed class FixedTime(DateTime utc) : TimeProvider
    {
        public override DateTimeOffset GetUtcNow() => new(utc, TimeSpan.Zero);
    }

    private (SqlHistorianConnection C, SqlServerSchemaBootstrapper B, SqlServerHistorianOptions O, string S) Make()
    {
        var s = SqlHistorianFixture.UniqueSchema();
        var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s };
        var c = new SqlHistorianConnection(o);
        return (c, new SqlServerSchemaBootstrapper(c, o), o, s);
    }

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

    private async Task<long> CountAsync(string sql)
    {
        await using var c = new SqlConnection(fx.ConnectionString);
        await c.OpenAsync();
        await using var cmd = c.CreateCommand();
        cmd.CommandText = sql;
        return Convert.ToInt64(await cmd.ExecuteScalarAsync());
    }

    /// <summary>3 tags x (5 old + 2 new) samples, plus 4 old + 1 new events. Retention 30 days at <see cref="Now"/>.</summary>
    private async Task SeedAsync(SqlServerSchemaBootstrapper b, string s)
    {
        await b.EnsureAsync(CancellationToken.None);
        var old = Now.AddDays(-40).ToString("yyyy-MM-ddTHH:mm:ss");
        var fresh = Now.AddDays(-1).ToString("yyyy-MM-ddTHH:mm:ss");
        var sql = new System.Text.StringBuilder();
        for (var t = 1; t <= 3; t++)
        {
            sql.Append($"INSERT [{s}].Tag (Name, DataType, CreatedUtc, UpdatedUtc) VALUES (N'tag{t}', 1, SYSUTCDATETIME(), SYSUTCDATETIME());");
        }
        sql.Append($"DECLARE @i int = 0; WHILE @i < 5 BEGIN INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, DATEADD(minute, @i, '{old}'), 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag; SET @i += 1; END;");
        sql.Append($"SET @i = 0; WHILE @i < 2 BEGIN INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, DATEADD(minute, @i, '{fresh}'), 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag; SET @i += 1; END;");
        sql.Append($"SET @i = 0; WHILE @i < 4 BEGIN INSERT [{s}].Event (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, EventTimeUtc, ReceivedTimeUtc) VALUES (CONCAT('old', @i), 's', 'a', 'n', 't', 1, 'Activated', DATEADD(minute, @i, '{old}'), SYSUTCDATETIME()); SET @i += 1; END;");
        sql.Append($"INSERT [{s}].Event (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, EventTimeUtc, ReceivedTimeUtc) VALUES ('new0', 's', 'a', 'n', 't', 1, 'Activated', '{fresh}', SYSUTCDATETIME());");
        await ExecAsync(sql.ToString());
    }

    [Theory]
    [InlineData(4000)]
    [InlineData(1)]   // forces the delete loop to iterate once per row
    public async Task Purge_removes_only_rows_older_than_the_window(int batch)
    {
        fx.SkipIfUnavailable();
        var (c, b, _, s) = Make();
        await SeedAsync(b, s);

        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, batch, time: new FixedTime(Now));
        var results = await purger.PurgeAsync(CancellationToken.None);

        results.ShouldBe([new RetentionPurgeResult("Sample", 15), new RetentionPurgeResult("Event", 4)]);
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(6);
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample WHERE SourceTimeUtc < '{Now.AddDays(-30):yyyy-MM-ddTHH:mm:ss}'")).ShouldBe(0);
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Event")).ShouldBe(1);
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag")).ShouldBe(3);
    }

    [Fact]
    public async Task A_second_purge_is_a_no_op()
    {
        fx.SkipIfUnavailable();
        var (c, b, _, s) = Make();
        await SeedAsync(b, s);
        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
        await purger.PurgeAsync(CancellationToken.None);
        var again = await purger.PurgeAsync(CancellationToken.None);
        again.ShouldBe([new RetentionPurgeResult("Sample", 0), new RetentionPurgeResult("Event", 0)]);
    }

    [Fact]
    public async Task A_purger_skips_while_another_holds_the_applock()
    {
        fx.SkipIfUnavailable();
        var (c, b, _, s) = Make();
        await SeedAsync(b, s);

        // Hold the same session applock the purger takes, on a separate connection.
        await using var holder = new SqlConnection(fx.ConnectionString);
        await holder.OpenAsync();
        await using (var cmd = holder.CreateCommand())
        {
            cmd.CommandText = $"EXEC sp_getapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0;";
            await cmd.ExecuteNonQueryAsync();
        }

        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
        (await purger.PurgeAsync(CancellationToken.None)).ShouldBeEmpty();
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(21);

        await using (var cmd = holder.CreateCommand())
        {
            cmd.CommandText = $"EXEC sp_releaseapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockOwner = 'Session';";
            await cmd.ExecuteNonQueryAsync();
        }
        (await purger.PurgeAsync(CancellationToken.None)).ShouldNotBeEmpty();
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(6);
    }

    /// <summary>True when a fresh, non-pooled session can take the retention applock (i.e. nobody holds it).</summary>
    private async Task<bool> LockIsFreeAsync(string s)
    {
        var cs = new SqlConnectionStringBuilder(fx.ConnectionString) { Pooling = false }.ConnectionString;
        await using var c = new SqlConnection(cs);
        await c.OpenAsync();
        await using var cmd = c.CreateCommand();
        cmd.CommandText = $"DECLARE @r int; EXEC @r = sp_getapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT @r;";
        return Convert.ToInt32(await cmd.ExecuteScalarAsync()) >= 0;
    }

    [Fact]
    public async Task The_applock_is_released_after_a_purge()
    {
        fx.SkipIfUnavailable();
        var (c, b, _, s) = Make();
        await SeedAsync(b, s);
        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
        await purger.PurgeAsync(CancellationToken.None);
        // A pooled-session re-entry would hide a missing release; a separate non-pooled session cannot.
        (await LockIsFreeAsync(s)).ShouldBeTrue();
    }

    [Fact]
    public async Task Cancelling_between_batches_stops_the_purge_and_still_releases_the_applock()
    {
        fx.SkipIfUnavailable();
        var (c, b, _, s) = Make();
        await SeedAsync(b, s);
        using var cts = new CancellationTokenSource();
        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 1, time: new FixedTime(Now));
        var batches = 0;
        purger.AfterBatch = () => { if (++batches == 2) cts.Cancel(); };

        await Should.ThrowAsync<OperationCanceledException>(() => purger.PurgeAsync(cts.Token));

        batches.ShouldBe(2);
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(19); // 21 - two batches of 1
        (await LockIsFreeAsync(s)).ShouldBeTrue();
    }

    [Fact]
    public async Task A_throwing_purge_releases_the_applock()
    {
        fx.SkipIfUnavailable();
        var (c, b, _, s) = Make();
        await SeedAsync(b, s);
        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 1, time: new FixedTime(Now));
        purger.AfterBatch = () => throw new InvalidOperationException("boom");
        await Should.ThrowAsync<InvalidOperationException>(() => purger.PurgeAsync(CancellationToken.None));
        (await LockIsFreeAsync(s)).ShouldBeTrue();
    }

    private sealed class CapturingLogger : Microsoft.Extensions.Logging.ILogger
    {
        public readonly List<Microsoft.Extensions.Logging.LogLevel> Levels = [];
        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
        public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
        public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId,
            TState state, Exception? exception, Func<TState, Exception?, string> formatter)
        {
            lock (Levels) Levels.Add(logLevel);
        }
    }

    [Fact]
    public async Task A_skipped_pass_logs_no_warning_and_does_not_clear_the_pool()
    {
        fx.SkipIfUnavailable();
        var (c, b, _, s) = Make();
        await SeedAsync(b, s);

        await using var holder = new SqlConnection(fx.ConnectionString);
        await holder.OpenAsync();
        await using (var cmd = holder.CreateCommand())
        {
            cmd.CommandText = $"EXEC sp_getapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0;";
            await cmd.ExecuteNonQueryAsync();
        }

        var log = new CapturingLogger();
        var clears = 0;
        var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, log, new FixedTime(Now))
        {
            ClearPool = _ => Interlocked.Increment(ref clears),
        };

        for (var i = 0; i < 3; i++) (await purger.PurgeAsync(CancellationToken.None)).ShouldBeEmpty();

        lock (log.Levels) log.Levels.ShouldNotContain(l => l >= Microsoft.Extensions.Logging.LogLevel.Warning);
        clears.ShouldBe(0);
        // The shared pool is still healthy.
        (await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(21);
    }
}