lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianRetention.cs

using System.Data;
using Microsoft.Data.SqlClient;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;

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

/// <summary>
///     Deletes <c>hist.Sample</c> / <c>hist.SampleModified</c> / <c>hist.Event</c> rows older than the retention window, in small
///     <c>DELETE TOP(n)</c> batches (under the lock-escalation threshold). Samples are purged per tag so each
///     delete is a seek on the clustered key <c>(TagId, SourceTimeUtc)</c>; events seek on <c>CX_Event_Time</c>.
///     A session <c>sp_getapplock</c> (timeout 0) serialises purgers that share one database; a loser skips.
///     To purge another table later, add it to <see cref="PurgeAsync"/>. On a v1 schema (no Part 11 trail, Gitea #674)
///     the <c>SampleModified</c> purge is skipped and absent from the results.
/// </summary>
internal sealed class SqlServerHistorianRetention : IHistorianRetention
{
    private readonly SqlHistorianConnection _conn;
    private readonly SqlServerSchemaBootstrapper _bootstrap;
    private readonly int _days;
    private readonly int _batch;
    private readonly TimeProvider _time;
    private readonly ILogger _log;

    public SqlServerHistorianRetention(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap,
        TimeSpan interval, int retentionDays, int batchSize, ILogger? log = null, TimeProvider? time = null)
    {
        _conn = conn;
        _bootstrap = bootstrap;
        Interval = interval;
        _days = retentionDays;
        _batch = batchSize;
        _log = log ?? NullLogger.Instance;
        _time = time ?? TimeProvider.System;
    }

    public TimeSpan Interval { get; }

    /// <summary>Test hook invoked after every delete batch.</summary>
    internal Action? AfterBatch { get; set; }

    /// <summary>Test seam for <see cref="SqlConnection.ClearPool"/>.</summary>
    internal Action<SqlConnection> ClearPool { get; set; } = SqlConnection.ClearPool;

    /// <summary>Test seam: the release statement. Tests point it at a failing statement to exercise the pool-clear fallback.</summary>
    internal string ReleaseSql { get; set; } = "EXEC sp_releaseapplock @Resource = @res, @LockOwner = 'Session';";

    public async Task<IReadOnlyList<RetentionPurgeResult>> PurgeAsync(CancellationToken ct)
    {
        await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
        var cutoff = _time.GetUtcNow().UtcDateTime.AddDays(-_days);

        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
        var acquired = false;
        try
        {
            try
            {
                acquired = await TryLockAsync(conn, ct).ConfigureAwait(false);
            }
            catch
            {
                // Ambiguous: the server may have granted the lock before the client saw the reply. We cannot
                // release what we are not sure we hold, so close the session physically instead.
                ClearPool(conn);
                throw;
            }

            if (!acquired)
            {
                _log.LogInformation("Historian retention purge skipped: another purger holds the applock.");
                return [];
            }

            // #674: a v1 schema has no SampleModified table — skip its purge rather than fail the whole pass.
            var trail = _bootstrap.SupportsModifiedTrail;
            long samples = 0, modified = 0;
            var tagIds = new List<int>();
            await using (var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")}"))
            await using (var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false))
                while (await r.ReadAsync(ct).ConfigureAwait(false)) tagIds.Add(r.GetInt32(0));

            foreach (var id in tagIds)
            {
                samples += await DeleteLoopAsync(conn,
                    $"DELETE TOP (@n) FROM {_conn.Q("Sample")} WHERE TagId = @id AND SourceTimeUtc < @cutoff",
                    cutoff, id, ct).ConfigureAwait(false);
                // The Part 11 modification trail (#668) ages out on the SAME source-time cutoff as the samples it
                // describes; a purge never writes to it (no trigger, no OUTPUT INTO).
                if (trail)
                {
                    modified += await DeleteLoopAsync(conn,
                        $"DELETE TOP (@n) FROM {_conn.Q("SampleModified")} WHERE TagId = @id AND SourceTimeUtc < @cutoff",
                        cutoff, id, ct).ConfigureAwait(false);
                }
            }

            var events = await DeleteLoopAsync(conn,
                $"DELETE TOP (@n) FROM {_conn.Q("Event")} WHERE EventTimeUtc < @cutoff",
                cutoff, null, ct).ConfigureAwait(false);

            return trail
                ?
                [
                    new RetentionPurgeResult("Sample", samples),
                    new RetentionPurgeResult("SampleModified", modified),
                    new RetentionPurgeResult("Event", events),
                ]
                :
                [
                    new RetentionPurgeResult("Sample", samples),
                    new RetentionPurgeResult("Event", events),
                ];
        }
        finally
        {
            // Only when WE hold it: releasing an applock we were refused fails, which would log a false Warning
            // and clear the shared pool on every skipped interval.
            if (acquired) await ReleaseAsync(conn).ConfigureAwait(false);
        }
    }

    private async Task<long> DeleteLoopAsync(SqlConnection conn, string sql, DateTime cutoff, int? tagId, CancellationToken ct)
    {
        long total = 0;
        while (true)
        {
            ct.ThrowIfCancellationRequested();
            await using var cmd = _conn.CreateCommand(conn, sql);
            cmd.Parameters.Add(new SqlParameter("@n", SqlDbType.Int) { Value = _batch });
            if (tagId is { } id) cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
            SqlHistorianConnection.AddDateTime2(cmd, "@cutoff", cutoff);
            var n = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
            total += n;
            AfterBatch?.Invoke();
            if (n < _batch) return total;
        }
    }

    private string LockResource => $"OtOpcUa.Historian.Retention.{_conn.Schema}";

    private async Task<bool> TryLockAsync(SqlConnection conn, CancellationToken ct)
    {
        await using var cmd = _conn.CreateCommand(conn,
            "DECLARE @r int; EXEC @r = sp_getapplock @Resource = @res, @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT @r;");
        cmd.Parameters.Add(new SqlParameter("@res", SqlDbType.NVarChar, 255) { Value = LockResource });
        return Convert.ToInt32(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)) >= 0;
    }

    private async Task ReleaseAsync(SqlConnection conn)
    {
        try
        {
            await using var cmd = _conn.CreateCommand(conn, ReleaseSql);
            cmd.Parameters.Add(new SqlParameter("@res", SqlDbType.NVarChar, 255) { Value = LockResource });
            await cmd.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            // A pooled connection keeps its server session (sp_reset_connection does not release a Session-owned
            // applock); clearing the pool makes the dispose physically close it, which releases the lock.
            _log.LogWarning("sp_releaseapplock failed ({Exception}); discarding the connection pool so the session " +
                "holding the retention applock is closed rather than pooled.", ex.GetType().Name);
            ClearPool(conn);
        }
    }
}