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

using Microsoft.Data.SqlClient;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

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

/// <summary>
///     The one place that maps value kinds to <c>Tag.DataType</c> (the schema stores a <see cref="DriverDataType"/> byte).
///     Used only when the writer has to create a tag row on a miss; provisioning later corrects it.
/// </summary>
internal static class SqlHistorianDataTypes
{
    /// <summary>Data type used when a tag is created from a batch that holds no typed value.</summary>
    public const DriverDataType DefaultForEmpty = DriverDataType.Float64;

    /// <summary>Maps a value kind to the stored <c>Tag.DataType</c> byte (Empty ⇒ <see cref="DefaultForEmpty"/>).</summary>
    /// <param name="kind">Value kind.</param>
    /// <returns>The <see cref="DriverDataType"/> as a byte.</returns>
    public static byte FromKind(HistorianValueKind kind) => (byte)(kind switch
    {
        HistorianValueKind.Double => DriverDataType.Float64,
        HistorianValueKind.Int64 => DriverDataType.Int64,
        HistorianValueKind.Boolean => DriverDataType.Boolean,
        HistorianValueKind.String => DriverDataType.String,
        HistorianValueKind.DateTime => DriverDataType.DateTime,
        _ => DefaultForEmpty,
    });
}

/// <summary>
///     <see cref="IHistorianValueWriter"/> over the SQL <c>Sample</c> table. Redelivery-idempotent: rows are
///     inserted only where <c>(TagId, SourceTimeUtc)</c> is absent, so the recorder's at-least-once redelivery
///     leaves the first-stored sample untouched. Never throws; returns false on any failure so the recorder keeps
///     the entries and its #538 per-tag quarantine keeps working.
/// </summary>
/// <remarks>
///     <b>Non-finite doubles.</b> NaN / ±Infinity cannot live in a SQL <c>float</c> (the TVP throws). They are stored
///     as <see cref="HistorianValueKind.Empty"/> keeping the original status code, so the sample's presence and
///     status survive and the tag is not quarantined for one bad reading.
///     <b>Duplicate timestamps within a batch</b> collapse to the first (matching insert-if-absent across batches).
/// </remarks>
internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter
{
    private readonly SqlHistorianConnection _conn;
    private readonly SqlServerSchemaBootstrapper _bootstrap;
    private readonly SqlServerTagCatalog _catalog;
    private readonly ILogger _log;
    private readonly Func<DateTime> _utcNow;

    /// <summary>Creates the writer.</summary>
    /// <param name="conn">Connection helper.</param>
    /// <param name="bootstrap">Schema bootstrapper.</param>
    /// <param name="catalog">Tag catalog.</param>
    /// <param name="log">Optional logger.</param>
    /// <param name="utcNow">Clock for null timestamps (tests).</param>
    public SqlServerHistorianValueWriter(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap,
        SqlServerTagCatalog catalog, ILogger? log = null, Func<DateTime>? utcNow = null)
    {
        _conn = conn;
        _bootstrap = bootstrap;
        _catalog = catalog;
        _log = log ?? NullLogger.Instance;
        _utcNow = utcNow ?? (() => DateTime.UtcNow);
    }

    /// <inheritdoc/>
    public async Task<bool> WriteLiveValuesAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
    {
        try
        {
            await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
            if (values is null || values.Count == 0) return true;

            var rows = Prepare(values, _utcNow);
            var dataType = SqlHistorianDataTypes.FromKind(rows.Select(r => r.Value.Kind).FirstOrDefault(k => k != HistorianValueKind.Empty));
            var tagId = await _catalog.GetOrCreateTagIdAsync(tag, dataType, ct).ConfigureAwait(false);

            var sql =
                $"INSERT INTO {_conn.Q("Sample")} (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " +
                "SELECT @tagId, r.SourceTimeUtc, r.ValueKind, r.NumValue, r.IntValue, r.StrValue, r.StatusCode, SYSUTCDATETIME() " +
                "FROM @rows r WHERE NOT EXISTS (SELECT 1 FROM " + _conn.Q("Sample") +
                " s WITH (UPDLOCK, HOLDLOCK) WHERE s.TagId = @tagId AND s.SourceTimeUtc = r.SourceTimeUtc)";

            // A duplicate-key error rolls the WHOLE statement back, so "someone else wrote a row" is NOT proof our rows are
            // present. The retry re-evaluates NOT EXISTS against what is now there; only its success counts.
            for (var attempt = 0; ; attempt++)
            {
                try
                {
                    await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
                    await using var cmd = _conn.CreateCommand(conn, sql);
                    cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId });
                    _conn.AddStructured(cmd, "@rows", "SampleRow", SampleRowMapper.ToSampleRecords(rows));
                    await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
                    return true;
                }
                catch (SqlException ex) when (attempt < 2 && SqlHistorianConnection.IsDuplicateKey(ex))
                {
                }
            }
        }
        catch (Exception ex)
        {
            _log.LogWarning("SQL historian live-value write failed ({Exception}); will retry.", ex.GetType().Name);
            return false;
        }
    }

    /// <summary>
    ///     Normalises a batch for the TVP: null timestamp ⇒ now, non-finite doubles ⇒ Empty (status kept), and
    ///     duplicate timestamps collapsed to the first.
    /// </summary>
    /// <param name="values">Input samples, in append order.</param>
    /// <param name="utcNow">Clock for null timestamps.</param>
    /// <returns>Time-unique rows.</returns>
    internal static List<SqlSampleRow> Prepare(IReadOnlyList<HistorizationValue> values, Func<DateTime> utcNow)
    {
        var seen = new HashSet<long>();
        var rows = new List<SqlSampleRow>(values.Count);
        foreach (var v in values)
        {
            var t = v.TimestampUtc ?? utcNow();
            t = t.Kind == DateTimeKind.Local ? t.ToUniversalTime() : DateTime.SpecifyKind(t, DateTimeKind.Utc);
            if (!seen.Add(t.Ticks)) continue;
            var value = v.Value;
            if (value.Kind == HistorianValueKind.Double && !double.IsFinite(value.Number))
                value = HistorianValue.Empty;
            rows.Add(new SqlSampleRow(t, value, v.StatusCode));
        }

        return rows;
    }
}