using Microsoft.Data.SqlClient;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// Status codes owned by the updater (named consts so StatusCodeParityTests sees them).
internal static class SqlHistorianUpdateStatusCodes
{
///
/// A value that cannot be stored: a Double NaN / ±Infinity, an Empty (null) value with a non-Bad status, or a
/// tag name longer than .
///
public const uint BadInvalidArgument = 0x80AB0000u;
}
///
/// 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.
/// Duplicate timestamps within one request (compared at full 100 ns tick precision):
/// Insert — the first occurrence is applied, later ones are BadEntryExists; 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 BadNoEntryExists.
/// Invalid items. A null or timestamp is
/// BadInvalidTimestamp. A Double NaN/±Infinity cannot live in the SQL float column and is
/// rejected with BadInvalidArgument (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
/// BadInvalidArgument, because a stored sample is never served better than "no value". Other items
/// still proceed.
/// Tag names over characters are refused, never
/// truncated: every item of the call (and DeleteRaw's single result) reports BadInvalidArgument and nothing
/// is written.
/// Unknown tag. Insert/Update create it via ; Replace and the
/// deletes report BadNoEntryExists (DeleteRaw: BadNoData) without creating anything.
///
internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
{
private readonly SqlHistorianConnection _conn;
private readonly SqlServerTagCatalog _tags;
/// Creates the updater.
/// Connection helper.
/// Tag catalog (runs the schema bootstrap itself).
public SqlServerHistorianDataUpdater(SqlHistorianConnection conn, SqlServerTagCatalog tags)
{
_conn = conn;
_tags = tags;
}
///
public async Task> UpdateDataAsync(string tag, HistoryUpdateMode mode,
IReadOnlyList values, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(values);
var results = new uint[values.Count];
if (!SqlServerTagCatalog.IsValidName(tag))
{
Array.Fill(results, SqlHistorianUpdateStatusCodes.BadInvalidArgument);
return results;
}
var winner = new Dictionary(); // tick -> index of the item actually sent
for (var i = 0; i < values.Count; i++)
{
var v = values[i];
if (v.TimestampUtc is not { } ts || ts == DateTime.MinValue)
{
results[i] = SqlHistorianStatusCodes.BadInvalidTimestamp;
continue;
}
if ((v.Value.Kind == HistorianValueKind.Double && !double.IsFinite(v.Value.Number))
|| (v.Value.Kind == HistorianValueKind.Empty && !StatusBits.IsBad(v.StatusCode)))
{
results[i] = SqlHistorianUpdateStatusCodes.BadInvalidArgument;
continue;
}
if (!winner.TryGetValue(ts.Ticks, out _))
winner[ts.Ticks] = i;
else if (mode == HistoryUpdateMode.Insert)
results[i] = SqlHistorianStatusCodes.BadEntryExists; // first occurrence stays the winner
else
winner[ts.Ticks] = i; // last wins
}
var send = winner.Values.OrderBy(i => i).ToList();
if (send.Count == 0) return results;
var tagId = mode == HistoryUpdateMode.Replace
? await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false)
: await _tags.GetOrCreateTagIdAsync(tag, SqlHistorianDataTypes.FromKind(values[send[0]].Value.Kind), ct).ConfigureAwait(false);
var outcome = tagId is null
? new Dictionary()
: await MergeAsync(tagId.Value, mode, send.Select(i => new SqlSampleRow(
values[i].TimestampUtc!.Value, values[i].Value, values[i].StatusCode)).ToList(), ct).ConfigureAwait(false);
// No OUTPUT row means the MERGE's absent branch: Insert => matched (exists); Replace => not matched.
var absent = mode == HistoryUpdateMode.Replace
? SqlHistorianStatusCodes.BadNoEntryExists
: SqlHistorianStatusCodes.BadEntryExists;
foreach (var i in send)
results[i] = outcome.TryGetValue(values[i].TimestampUtc!.Value.Ticks, out var code) ? code : absent;
// Superseded earlier duplicates (Replace/Update) share the winner's code.
if (mode != HistoryUpdateMode.Insert)
for (var i = 0; i < values.Count; i++)
if (values[i].TimestampUtc is { } ts && results[i] == 0
&& winner.TryGetValue(ts.Ticks, out var w) && w != i)
results[i] = results[w];
return results;
}
///
public async Task DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
if (!SqlServerTagCatalog.IsValidName(tag)) return SqlHistorianUpdateStatusCodes.BadInvalidArgument;
var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
if (tagId is null) return SqlHistorianStatusCodes.BadNoData;
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")} WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
SqlHistorianConnection.AddDateTime2(cmd, "@s", startUtc);
SqlHistorianConnection.AddDateTime2(cmd, "@e", endUtc);
rows = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
}
await tx.CommitAsync(ct).ConfigureAwait(false);
return rows > 0 ? SqlHistorianStatusCodes.Good : SqlHistorianStatusCodes.BadNoData;
}
///
public async Task> DeleteAtTimeAsync(string tag, IReadOnlyList timesUtc, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(timesUtc);
var results = new uint[timesUtc.Count];
if (!SqlServerTagCatalog.IsValidName(tag))
{
Array.Fill(results, SqlHistorianUpdateStatusCodes.BadInvalidArgument);
return results;
}
var first = new Dictionary();
for (var i = 0; i < timesUtc.Count; i++)
{
results[i] = SqlHistorianStatusCodes.BadNoEntryExists;
if (timesUtc[i] == DateTime.MinValue) results[i] = SqlHistorianStatusCodes.BadInvalidTimestamp;
else first.TryAdd(timesUtc[i].Ticks, i);
}
if (first.Count == 0) return results;
var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
if (tagId is null) return results;
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 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 });
_conn.AddStructured(cmd, "@times", "TimeRow",
SampleRowMapper.ToTimeRecords(first.Keys.Select(t => new DateTime(t, DateTimeKind.Utc))));
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await rd.ReadAsync(ct).ConfigureAwait(false))
if (first.TryGetValue(SqlHistorianConnection.GetUtc(rd, 0).Ticks, out var idx))
results[idx] = SqlHistorianStatusCodes.Good;
}
await tx.CommitAsync(ct).ConfigureAwait(false);
return results;
}
private async Task> MergeAsync(int tagId, HistoryUpdateMode mode,
List rows, 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()) ";
sql += "OUTPUT $action, r.SourceTimeUtc;";
var outcome = new Dictionary();
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 });
_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"
? SqlHistorianStatusCodes.GoodEntryInserted
: SqlHistorianStatusCodes.GoodEntryReplaced;
}
await tx.CommitAsync(ct).ConfigureAwait(false);
return outcome;
}
}