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

using Microsoft.Data.SqlClient;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

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

/// <summary>
///     OPC UA HistoryUpdate against the SQL Server sample table (spec §4 "Updates"). One transaction per call;
///     exactly one status per input item, in input order.
///     <para><b>Duplicate timestamps within one request</b> (compared at full 100 ns tick precision):
///     Insert — the first occurrence is applied, later ones are <c>BadEntryExists</c>; Replace/Update — last wins,
///     and every earlier occurrence of that timestamp reports the same code as the winning one.
///     DeleteAtTime — the first occurrence reports the real outcome, later ones <c>BadNoEntryExists</c>.</para>
///     <para><b>Invalid items.</b> A null or <see cref="DateTime.MinValue"/> timestamp is
///     <c>BadInvalidTimestamp</c>. A Double NaN/±Infinity cannot live in the SQL <c>float</c> column and is
///     rejected with <c>BadInvalidArgument</c> (not silently stored as Empty). An Empty (null) value is only
///     accepted with a Bad status (an operator-authored outage marker); with a Good or Uncertain status it is
///     <c>BadInvalidArgument</c>, because a stored sample is never served better than "no value". Other items
///     still proceed.</para>
///     <para><b>Tag names over <see cref="HistorianTagNames.MaxLength"/> characters</b> are refused, never
///     truncated: every item of the call (and DeleteRaw's single result) reports <c>BadInvalidArgument</c> and nothing
///     is written.</para>
///     <para><b>Unknown tag.</b> Insert/Update create it via <see cref="SqlServerTagCatalog"/>; Replace and the
///     deletes report <c>BadNoEntryExists</c> (DeleteRaw: <c>BadNoData</c>) without creating anything.</para>
///     <para><b>Modified-value history (Part 11, #668).</b> Every row the MERGE inserts or updates and every row a
///     delete removes is copied into <c>SampleModified</c> by an <c>OUTPUT … INTO</c> on the SAME statement — the prior
///     value from <c>deleted.*</c> (NULL for an insert), the caller's <see cref="HistoryUpdateContext"/>, the
///     <see cref="HistoryModificationType"/> and <c>SYSUTCDATETIME()</c>. One statement means the sample change and its
///     audit row commit or roll back together: a failed update leaves no audit row.</para>
///     <para><b>Schema v1 (Gitea #674).</b> Without the trail (<see cref="SqlServerSchemaBootstrapper.SupportsModifiedTrail"/>
///     false) a change could not be audited, so every call throws <see cref="NotSupportedException"/> — which the node
///     manager reports as a node-level <c>BadHistoryOperationUnsupported</c>, on all three arms alike — and nothing is
///     sent to the database beyond the one-time schema check.</para>
/// </summary>
internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
{
    private readonly SqlHistorianConnection _conn;
    private readonly SqlServerTagCatalog _tags;
    private readonly SqlServerSchemaBootstrapper _bootstrap;

    /// <summary>Creates the updater.</summary>
    /// <param name="conn">Connection helper.</param>
    /// <param name="bootstrap">Schema bootstrapper — decides whether the modification trail exists (#674).</param>
    /// <param name="tags">Tag catalog (runs the schema bootstrap itself).</param>
    public SqlServerHistorianDataUpdater(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap,
        SqlServerTagCatalog tags)
    {
        _conn = conn;
        _bootstrap = bootstrap;
        _tags = tags;
    }

    /// <summary>
    ///     #674: ensures the schema and refuses the call when it has no Part 11 trail (no SQL after the first success).
    /// </summary>
    /// <param name="ct">Cancellation.</param>
    /// <returns>A task completing when the trail is available.</returns>
    /// <exception cref="NotSupportedException">The schema is below <see cref="SqlServerSchemaBootstrapper.ModifiedTrailVersion"/>.</exception>
    private async Task RequireTrailAsync(CancellationToken ct)
    {
        await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
        if (!_bootstrap.SupportsModifiedTrail)
            throw new NotSupportedException(
                $"Historian schema [{_conn.Schema}] is below version {SqlServerSchemaBootstrapper.ModifiedTrailVersion}: " +
                "HistoryUpdate needs the Part 11 modification trail (apply 002_sample_modified.sql and restart).");
    }

    /// <inheritdoc/>
    public async Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
        IReadOnlyList<HistorizationValue> values, HistoryUpdateContext context, CancellationToken ct)
    {
        ArgumentNullException.ThrowIfNull(tag);
        ArgumentNullException.ThrowIfNull(values);
        ArgumentNullException.ThrowIfNull(context);
        await RequireTrailAsync(ct).ConfigureAwait(false);
        var plan = HistoryUpdatePlanner.Plan(tag, mode, values);
        if (plan.Winners.Count == 0) return HistoryUpdatePlanner.Complete(plan, mode, new Dictionary<long, UpsertOutcome>());

        var tagId = mode == HistoryUpdateMode.Replace
            ? await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false)
            : await _tags.GetOrCreateTagIdAsync(tag, HistorianDataTypes.FromKind(plan.Winners[0].Row.Value.Kind), ct).ConfigureAwait(false);

        var outcome = tagId is null
            ? new Dictionary<long, UpsertOutcome>()
            : await MergeAsync(tagId.Value, mode, plan.Winners.Select(w => w.Row).ToList(), context, ct).ConfigureAwait(false);
        return HistoryUpdatePlanner.Complete(plan, mode, outcome);
    }

    /// <inheritdoc/>
    public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, HistoryUpdateContext context,
        CancellationToken ct)
    {
        ArgumentNullException.ThrowIfNull(tag);
        ArgumentNullException.ThrowIfNull(context);
        await RequireTrailAsync(ct).ConfigureAwait(false);
        if (!HistorianTagNames.IsValid(tag)) return HistorianStatusCodes.BadInvalidArgument;
        var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
        if (tagId is null) return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0);
        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
        await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
        int rows;
        await using (var cmd = _conn.CreateCommand(conn,
                         $"DELETE FROM {_conn.Q("Sample")} {AuditInto(HistoryModificationType.Delete)} " +
                         "WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
        {
            cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
            AddAuditParameters(cmd, context);
            SqlHistorianConnection.AddDateTime2(cmd, "@s", startUtc);
            SqlHistorianConnection.AddDateTime2(cmd, "@e", endUtc);
            rows = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
        }
        await tx.CommitAsync(ct).ConfigureAwait(false);
        return HistoryUpdatePlanner.DeleteRawResult(tagKnown: true, rows);
    }

    /// <inheritdoc/>
    public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc,
        HistoryUpdateContext context, CancellationToken ct)
    {
        ArgumentNullException.ThrowIfNull(tag);
        ArgumentNullException.ThrowIfNull(timesUtc);
        ArgumentNullException.ThrowIfNull(context);
        await RequireTrailAsync(ct).ConfigureAwait(false);
        if (!HistorianTagNames.IsValid(tag))
        {
            var refused = new uint[timesUtc.Count];
            Array.Fill(refused, HistorianStatusCodes.BadInvalidArgument);
            return refused;
        }
        var distinct = new HashSet<long>();
        foreach (var t in timesUtc)
            if (t != DateTime.MinValue) distinct.Add(t.Ticks);
        var deleted = new HashSet<long>();
        if (distinct.Count == 0) return HistoryUpdatePlanner.DeleteAtTimeResults(timesUtc, deleted, tagKnown: false);
        var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
        if (tagId is null) return HistoryUpdatePlanner.DeleteAtTimeResults(timesUtc, deleted, tagKnown: false);

        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
        await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
        await using (var cmd = _conn.CreateCommand(conn,
                         $"DELETE s {AuditInto(HistoryModificationType.Delete)} OUTPUT deleted.SourceTimeUtc " +
                         $"FROM {_conn.Q("Sample")} AS s JOIN @times AS t ON s.SourceTimeUtc = t.T WHERE s.TagId = @tagId", tx))
        {
            cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
            AddAuditParameters(cmd, context);
            _conn.AddStructured(cmd, "@times", "TimeRow",
                SampleRowMapper.ToTimeRecords(distinct.Select(t => new DateTime(t, DateTimeKind.Utc))));
            await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
            while (await rd.ReadAsync(ct).ConfigureAwait(false))
                deleted.Add(SqlHistorianConnection.GetUtc(rd, 0).Ticks);
        }
        await tx.CommitAsync(ct).ConfigureAwait(false);
        return HistoryUpdatePlanner.DeleteAtTimeResults(timesUtc, deleted, tagKnown: true);
    }

    private async Task<Dictionary<long, UpsertOutcome>> MergeAsync(int tagId, HistoryUpdateMode mode,
        List<HistorianSampleRow> rows, HistoryUpdateContext context, CancellationToken ct)
    {
        var sql = $"MERGE {_conn.Q("Sample")} WITH (HOLDLOCK) AS t USING @rows AS r ON t.TagId = @tagId AND t.SourceTimeUtc = r.SourceTimeUtc ";
        if (mode != HistoryUpdateMode.Insert)
            sql += "WHEN MATCHED THEN UPDATE SET ValueKind = r.ValueKind, NumValue = r.NumValue, IntValue = r.IntValue, " +
                   "StrValue = r.StrValue, StatusCode = r.StatusCode, ServerTimeUtc = SYSUTCDATETIME() ";
        if (mode != HistoryUpdateMode.Replace)
            sql += "WHEN NOT MATCHED THEN INSERT (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " +
                   "VALUES (@tagId, r.SourceTimeUtc, r.ValueKind, r.NumValue, r.IntValue, r.StrValue, r.StatusCode, SYSUTCDATETIME()) ";
        // The MERGE's deleted.* is NULL for an inserted row, so one OUTPUT INTO records both an insert (no prior value)
        // and a replace (the prior value). TagId comes from inserted.* — present on both branches.
        sql += $"{AuditInto(ToModificationType(mode), "inserted")} OUTPUT $action, r.SourceTimeUtc;";

        var outcome = new Dictionary<long, UpsertOutcome>();
        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
        await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
        await using (var cmd = _conn.CreateCommand(conn, sql, tx))
        {
            cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId });
            AddAuditParameters(cmd, context);
            _conn.AddStructured(cmd, "@rows", "SampleRow", SampleRowMapper.ToSampleRecords(rows));
            await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
            while (await rd.ReadAsync(ct).ConfigureAwait(false))
                outcome[SqlHistorianConnection.GetUtc(rd, 1).Ticks] = rd.GetString(0) == "INSERT"
                    ? UpsertOutcome.Inserted
                    : UpsertOutcome.Replaced;
        }
        await tx.CommitAsync(ct).ConfigureAwait(false);
        return outcome;
    }

    /// <summary>The Part 11 modification type recorded for an update mode.</summary>
    /// <param name="mode">The update mode.</param>
    /// <returns>Insert ⇒ Insert, Replace ⇒ Replace, Update ⇒ Update.</returns>
    internal static HistoryModificationType ToModificationType(HistoryUpdateMode mode) => mode switch
    {
        HistoryUpdateMode.Insert => HistoryModificationType.Insert,
        HistoryUpdateMode.Replace => HistoryModificationType.Replace,
        _ => HistoryModificationType.Update,
    };

    /// <summary>
    ///     The <c>OUTPUT … INTO SampleModified</c> clause copying each affected row's prior value (<c>deleted.*</c>) into
    ///     the modification trail. <paramref name="keySource"/> names the pseudo-table that carries <c>TagId</c> /
    ///     <c>SourceTimeUtc</c> for every affected row (<c>deleted</c> for a DELETE, <c>inserted</c> for the MERGE).
    /// </summary>
    private string AuditInto(HistoryModificationType type, string keySource = "deleted") =>
        $"OUTPUT {keySource}.TagId, {keySource}.SourceTimeUtc, SYSUTCDATETIME(), {(byte)type}, @principal, @apiKeyId, " +
        "deleted.ValueKind, deleted.NumValue, deleted.IntValue, deleted.StrValue, deleted.StatusCode, deleted.ServerTimeUtc " +
        $"INTO {_conn.Q("SampleModified")} (TagId, SourceTimeUtc, ModificationTimeUtc, UpdateType, Principal, ApiKeyId, " +
        "ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc)";

    private static void AddAuditParameters(SqlCommand cmd, HistoryUpdateContext context)
    {
        cmd.Parameters.Add(new SqlParameter("@principal", System.Data.SqlDbType.NVarChar, HistoryUpdateContext.MaxIdentityLength)
            { Value = context.StoredPrincipal });
        cmd.Parameters.Add(new SqlParameter("@apiKeyId", System.Data.SqlDbType.NVarChar, HistoryUpdateContext.MaxIdentityLength)
            { Value = (object?)context.StoredApiKeyId ?? DBNull.Value });
    }
}