diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs new file mode 100644 index 00000000..4d7bdb1f --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerAlarmHistorianWriter.cs @@ -0,0 +1,168 @@ +using System.Data; +using System.Security.Cryptography; +using System.Text; +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; + +/// +/// over the SQL Event table. Per event: insert-if-absent keyed by a +/// deterministic , so the store-and-forward drain's at-least-once redelivery (and both +/// pair nodes writing the same event) converge on one row. Outcomes: Ack on insert or already-present; +/// RetryPlease on transient SQL / schema-bootstrap / unclassified SQL errors and on cancellation; PermanentFail only +/// for data errors (truncation, conversion, null, range) and non-SQL exceptions. Never throws. +/// +internal sealed class SqlServerAlarmHistorianWriter : IAlarmHistorianWriter +{ + // 2628/8152 string truncation, 515 null into NOT NULL, 245/8114 conversion, 220/8115 overflow, 242/241 datetime range. + private static readonly int[] DataErrorNumbers = [2628, 8152, 515, 245, 8114, 220, 8115, 242, 241]; + + private readonly SqlHistorianConnection _conn; + private readonly SqlServerSchemaBootstrapper _bootstrap; + private readonly ILogger _log; + + /// Creates the writer. + /// Connection helper. + /// Schema bootstrapper. + /// Optional logger. + public SqlServerAlarmHistorianWriter(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, ILogger? log = null) + { + _conn = conn; + _bootstrap = bootstrap; + _log = log ?? NullLogger.Instance; + } + + /// + public async Task> WriteBatchAsync(IReadOnlyList batch, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(batch); + if (batch.Count == 0) return []; + var outcomes = new HistorianWriteOutcome[batch.Count]; + try + { + await _bootstrap.EnsureAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _log.LogWarning("SQL alarm historian bootstrap failed ({Exception}); will retry.", ex.GetType().Name); + Array.Fill(outcomes, HistorianWriteOutcome.RetryPlease); + return outcomes; + } + + for (var i = 0; i < batch.Count; i++) + { + if (cancellationToken.IsCancellationRequested) { outcomes[i] = HistorianWriteOutcome.RetryPlease; continue; } + outcomes[i] = await WriteOneAsync(batch[i], cancellationToken).ConfigureAwait(false); + } + + return outcomes; + } + + private async Task WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct) + { + try + { + // String parameters are NVARCHAR(MAX) on purpose: a sized parameter silently truncates client-side, which would + // hide an over-long value; the server-side 8152 is what classifies it PermanentFail. + var sql = + $"INSERT INTO {_conn.Q("Event")} (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, UserName, Comment, EventTimeUtc, ReceivedTimeUtc) " + + "SELECT @id, @src, @aid, @an, @at, @sev, @ek, @msg, @usr, @cmt, @t, SYSUTCDATETIME() " + + $"WHERE NOT EXISTS (SELECT 1 FROM {_conn.Q("Event")} x WITH (UPDLOCK, HOLDLOCK) WHERE x.EventId = @id)"; + { + try + { + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, sql); + cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.NVarChar, 64) { Value = ComputeEventId(e) }); + cmd.Parameters.Add(new SqlParameter("@src", SqlDbType.NVarChar, -1) { Value = e.EquipmentPath ?? "" }); + cmd.Parameters.Add(new SqlParameter("@aid", SqlDbType.NVarChar, -1) { Value = e.AlarmId ?? "" }); + cmd.Parameters.Add(new SqlParameter("@an", SqlDbType.NVarChar, -1) { Value = e.AlarmName ?? "" }); + cmd.Parameters.Add(new SqlParameter("@at", SqlDbType.NVarChar, -1) { Value = e.AlarmTypeName ?? "" }); + cmd.Parameters.Add(new SqlParameter("@sev", SqlDbType.Int) { Value = ToNumericSeverity(e.Severity) }); + cmd.Parameters.Add(new SqlParameter("@ek", SqlDbType.NVarChar, -1) { Value = e.EventKind ?? "" }); + cmd.Parameters.Add(new SqlParameter("@msg", SqlDbType.NVarChar, -1) { Value = (object?)e.Message ?? DBNull.Value }); + cmd.Parameters.Add(new SqlParameter("@usr", SqlDbType.NVarChar, -1) { Value = (object?)e.User ?? DBNull.Value }); + cmd.Parameters.Add(new SqlParameter("@cmt", SqlDbType.NVarChar, -1) { Value = (object?)e.Comment ?? DBNull.Value }); + SqlHistorianConnection.AddDateTime2(cmd, "@t", e.TimestampUtc); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + return HistorianWriteOutcome.Ack; + } + // Single-row statement: a duplicate-key error means the row IS there (a concurrent writer won). + catch (SqlException ex) when (SqlHistorianConnection.IsDuplicateKey(ex)) + { + return HistorianWriteOutcome.Ack; + } + } + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return HistorianWriteOutcome.RetryPlease; + } + catch (Exception ex) + { + var outcome = Classify(ex); + _log.LogWarning("SQL alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome); + return outcome; + } + } + + private static HistorianWriteOutcome Classify(Exception ex) + { + if (SqlHistorianConnection.IsTransient(ex)) return HistorianWriteOutcome.RetryPlease; + if (ex is SqlException sql) + return sql.Errors.Cast().Any(x => Array.IndexOf(DataErrorNumbers, x.Number) >= 0) + ? HistorianWriteOutcome.PermanentFail + : HistorianWriteOutcome.RetryPlease; // config/auth/availability problems must not dead-letter data + return HistorianWriteOutcome.PermanentFail; + } + + // Same numeric Part 9 priorities the gateway path sends (AlarmEventMapper.ToNumericSeverity). + private static int ToNumericSeverity(AlarmSeverity s) => s switch + { + AlarmSeverity.Low => 200, + AlarmSeverity.Medium => 500, + AlarmSeverity.High => 700, + AlarmSeverity.Critical => 900, + _ => 500, + }; + + /// + /// Lowercase-hex SHA-256 over length-prefixed UTF-8 fields (little-endian int32 byte length, -1 for null): + /// EquipmentPath, AlarmId, EventKind, TimestampUtc.Ticks (decimal), User, Comment, Message. Length-prefixing keeps + /// field boundaries unambiguous ("a|b"+"c" ≠ "a"+"b|c"). + /// + /// The event. + /// 64 hex characters. + internal static string ComputeEventId(AlarmHistorianEvent e) + { + using var ms = new MemoryStream(); + Field(ms, e.EquipmentPath); + Field(ms, e.AlarmId); + Field(ms, e.EventKind); + Field(ms, e.TimestampUtc.Ticks.ToString(System.Globalization.CultureInfo.InvariantCulture)); + Field(ms, e.User); + Field(ms, e.Comment); + Field(ms, e.Message); + return Convert.ToHexStringLower(SHA256.HashData(ms.ToArray())); + } + + private static void Field(MemoryStream ms, string? value) + { + Span len = stackalloc byte[4]; + if (value is null) + { + System.Buffers.Binary.BinaryPrimitives.WriteInt32LittleEndian(len, -1); + ms.Write(len); + return; + } + + var bytes = Encoding.UTF8.GetBytes(value); + System.Buffers.Binary.BinaryPrimitives.WriteInt32LittleEndian(len, bytes.Length); + ms.Write(len); + ms.Write(bytes); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs new file mode 100644 index 00000000..c1e9582c --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs @@ -0,0 +1,134 @@ +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; + +/// +/// The one place that maps value kinds to Tag.DataType (the schema stores a byte). +/// Used only when the writer has to create a tag row on a miss; provisioning later corrects it. +/// +internal static class SqlHistorianDataTypes +{ + /// Data type used when a tag is created from a batch that holds no typed value. + public const DriverDataType DefaultForEmpty = DriverDataType.Float64; + + /// Maps a value kind to the stored Tag.DataType byte (Empty ⇒ ). + /// Value kind. + /// The as a byte. + 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, + }); +} + +/// +/// over the SQL Sample table. Redelivery-idempotent: rows are +/// inserted only where (TagId, SourceTimeUtc) 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. +/// +/// +/// Non-finite doubles. NaN / ±Infinity cannot live in a SQL float (the TVP throws). They are stored +/// as keeping the original status code, so the sample's presence and +/// status survive and the tag is not quarantined for one bad reading. +/// Duplicate timestamps within a batch collapse to the first (matching insert-if-absent across batches). +/// +internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter +{ + private readonly SqlHistorianConnection _conn; + private readonly SqlServerSchemaBootstrapper _bootstrap; + private readonly SqlServerTagCatalog _catalog; + private readonly ILogger _log; + private readonly Func _utcNow; + + /// Creates the writer. + /// Connection helper. + /// Schema bootstrapper. + /// Tag catalog. + /// Optional logger. + /// Clock for null timestamps (tests). + public SqlServerHistorianValueWriter(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, + SqlServerTagCatalog catalog, ILogger? log = null, Func? utcNow = null) + { + _conn = conn; + _bootstrap = bootstrap; + _catalog = catalog; + _log = log ?? NullLogger.Instance; + _utcNow = utcNow ?? (() => DateTime.UtcNow); + } + + /// + public async Task WriteLiveValuesAsync(string tag, IReadOnlyList 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; + } + } + + /// + /// Normalises a batch for the TVP: null timestamp ⇒ now, non-finite doubles ⇒ Empty (status kept), and + /// duplicate timestamps collapsed to the first. + /// + /// Input samples, in append order. + /// Clock for null timestamps. + /// Time-unique rows. + internal static List Prepare(IReadOnlyList values, Func utcNow) + { + var seen = new HashSet(); + var rows = new List(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; + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs new file mode 100644 index 00000000..c0582b88 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerTagProvisioner.cs @@ -0,0 +1,120 @@ +using System.Data; +using System.Text; +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; + +/// +/// for the SQL backend: one MERGE … WITH (HOLDLOCK) upsert per chunk of +/// tags (chunked only to stay under SQL's 2100-parameter limit). Every data type is provisionable (Skipped is +/// always 0). A null engineering unit / description in a request keeps the stored value rather than wiping it. +/// Never throws. +/// +internal sealed class SqlServerTagProvisioner : IHistorianProvisioning +{ + private const int ChunkSize = 400; // 4 parameters per tag + private const int MaxNameLength = 400; + private const int MaxUnitLength = 64; + private const int MaxDescriptionLength = 400; + private const int MaxAttempts = 4; + + private readonly SqlHistorianConnection _conn; + private readonly SqlServerSchemaBootstrapper _bootstrap; + private readonly ILogger _log; + + /// Creates the provisioner. + /// Connection helper. + /// Schema bootstrapper. + /// Optional logger. + public SqlServerTagProvisioner(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap, ILogger? log = null) + { + _conn = conn; + _bootstrap = bootstrap; + _log = log ?? NullLogger.Instance; + } + + /// + public async Task EnsureTagsAsync(IReadOnlyList requests, CancellationToken ct) + { + var requested = requests?.Count ?? 0; + if (requested == 0) return new HistorianProvisionResult(0, 0, 0, 0); + var failed = 0; + var ensured = 0; + try + { + await _bootstrap.EnsureAsync(ct).ConfigureAwait(false); + + // Distinct names (a MERGE source must not repeat a key); last request wins. Over-long/empty names fail alone. + var byName = new Dictionary(StringComparer.Ordinal); + foreach (var r in requests!) + { + if (r is null || string.IsNullOrWhiteSpace(r.TagName) || r.TagName.Length > MaxNameLength) { failed++; continue; } + byName[r.TagName] = (r, byName.TryGetValue(r.TagName, out var e) ? e.Count + 1 : 1); + } + + var ordered = byName.OrderBy(kv => kv.Key, StringComparer.Ordinal).Select(kv => kv.Value).ToList(); + for (var i = 0; i < ordered.Count; i += ChunkSize) + { + var chunk = ordered.Skip(i).Take(ChunkSize).ToList(); + var weight = chunk.Sum(c => c.Count); + if (await TryMergeAsync(chunk.Select(c => c.Req).ToList(), ct).ConfigureAwait(false)) ensured += weight; + else failed += weight; + } + } + catch (Exception ex) + { + _log.LogWarning("SQL historian provisioning failed ({Exception}).", ex.GetType().Name); + failed = requested - ensured; + } + + return new HistorianProvisionResult(requested, ensured, 0, failed); + } + + private async Task TryMergeAsync(List chunk, CancellationToken ct) + { + var sql = new StringBuilder(); + sql.Append($"MERGE {_conn.Q("Tag")} WITH (HOLDLOCK) AS t USING (VALUES "); + for (var i = 0; i < chunk.Count; i++) + sql.Append(i == 0 ? "" : ",").Append($"(@n{i},@d{i},@e{i},@c{i})"); + sql.Append(") AS s (Name, DataType, EngineeringUnit, Description) ON t.Name = s.Name " + + "WHEN MATCHED THEN UPDATE SET DataType = s.DataType, EngineeringUnit = COALESCE(s.EngineeringUnit, t.EngineeringUnit), " + + "Description = COALESCE(s.Description, t.Description), UpdatedUtc = SYSUTCDATETIME() " + + "WHEN NOT MATCHED THEN INSERT (Name, DataType, EngineeringUnit, Description, CreatedUtc, UpdatedUtc) " + + "VALUES (s.Name, s.DataType, s.EngineeringUnit, s.Description, SYSUTCDATETIME(), SYSUTCDATETIME());"); + + for (var attempt = 1; ; attempt++) + { + try + { + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, sql.ToString()); + for (var i = 0; i < chunk.Count; i++) + { + var r = chunk[i]; + cmd.Parameters.Add(new SqlParameter($"@n{i}", SqlDbType.NVarChar, MaxNameLength) { Value = r.TagName }); + cmd.Parameters.Add(new SqlParameter($"@d{i}", SqlDbType.TinyInt) { Value = (byte)r.DataType }); + cmd.Parameters.Add(new SqlParameter($"@e{i}", SqlDbType.NVarChar, MaxUnitLength) { Value = Clip(r.EngineeringUnit, MaxUnitLength) }); + cmd.Parameters.Add(new SqlParameter($"@c{i}", SqlDbType.NVarChar, MaxDescriptionLength) { Value = Clip(r.Description, MaxDescriptionLength) }); + } + + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + return true; + } + catch (SqlException ex) when (attempt < MaxAttempts && + (SqlHistorianConnection.IsDuplicateKey(ex) || SqlHistorianConnection.IsTransient(ex))) + { + await Task.Delay(20 * attempt, ct).ConfigureAwait(false); + } + catch (Exception ex) + { + _log.LogWarning("SQL historian tag MERGE failed ({Exception}).", ex.GetType().Name); + return false; + } + } + } + + private static object Clip(string? s, int max) => s is null ? DBNull.Value : s.Length <= max ? s : s[..max]; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AlarmWriterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AlarmWriterTests.cs new file mode 100644 index 00000000..36faa9fc --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/AlarmWriterTests.cs @@ -0,0 +1,89 @@ +using Microsoft.Data.SqlClient; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +[Collection(SqlHistorianCollection.Name)] +public sealed class AlarmWriterTests(SqlHistorianFixture fx) +{ + private static readonly DateTime T = new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1234567); + + private (SqlServerAlarmHistorianWriter W, string S) Make(string? connection = null, bool auto = true) + { + var s = SqlHistorianFixture.UniqueSchema(); + var o = new SqlServerHistorianOptions { ConnectionString = connection ?? fx.ConnectionString, Schema = s, AutoCreateSchema = auto }; + var c = new SqlHistorianConnection(o); + return (new SqlServerAlarmHistorianWriter(c, new SqlServerSchemaBootstrapper(c, o)), s); + } + + private static AlarmHistorianEvent Ev(string kind = "Activated", string type = "LimitAlarm", string? comment = "c") => + new("alarm-1", "Plant/Line/Eq", "HiTemp", type, AlarmSeverity.High, kind, "msg", "user", comment, T); + + [Fact] + public async Task Event_round_trips_and_a_repeat_is_one_row_with_both_Ack() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + var first = await w.WriteBatchAsync([Ev(), Ev()], CancellationToken.None); + first.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]); + (await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]); + (await w.WriteBatchAsync([Ev(kind: "Cleared")], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]); + + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = $"SELECT SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, UserName, Comment, EventTimeUtc, ReceivedTimeUtc FROM [{s}].Event WHERE EventKind='Activated'"; + await using var r = await cmd.ExecuteReaderAsync(); + (await r.ReadAsync()).ShouldBeTrue(); + r.GetString(0).ShouldBe("Plant/Line/Eq"); + r.GetString(1).ShouldBe("alarm-1"); + r.GetString(2).ShouldBe("HiTemp"); + r.GetString(3).ShouldBe("LimitAlarm"); + r.GetInt32(4).ShouldBe(700); + r.GetString(6).ShouldBe("msg"); + r.GetString(7).ShouldBe("user"); + r.GetString(8).ShouldBe("c"); + SqlHistorianConnection.GetUtc(r, 9).Ticks.ShouldBe(T.Ticks); + (DateTime.UtcNow - SqlHistorianConnection.GetUtc(r, 10)).ShouldBeLessThan(TimeSpan.FromMinutes(1)); + (await r.ReadAsync()).ShouldBeFalse(); + await r.CloseAsync(); + cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event"; + Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(2); + } + + [Fact] + public async Task Over_long_AlarmTypeName_is_PermanentFail_and_siblings_still_Ack() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", type: new string('t', 101)), Ev(kind: "C", comment: null)], CancellationToken.None); + r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack]); + } + + [Fact] + public async Task Concurrent_writers_of_the_same_event_all_Ack_with_one_row() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + await w.WriteBatchAsync([Ev(kind: "warm")], CancellationToken.None); + var rs = await Task.WhenAll(Enumerable.Range(0, 12).Select(_ => Task.Run(() => w.WriteBatchAsync([Ev()], CancellationToken.None)))); + rs.ShouldAllBe(x => x[0] == HistorianWriteOutcome.Ack); + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event WHERE EventKind='Activated'"; + Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(1); + } + + [Fact] + public async Task Unreachable_database_is_RetryPlease_not_an_exception_and_empty_batch_is_empty() + { + fx.SkipIfUnavailable(); + var (w, _) = Make(fx.ConnectionString + ";Connect Timeout=1", auto: false); // schema missing + (await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.RetryPlease]); + (await w.WriteBatchAsync([], CancellationToken.None)).ShouldBeEmpty(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs new file mode 100644 index 00000000..5ddff410 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/ProvisioningTests.cs @@ -0,0 +1,94 @@ +using Microsoft.Data.SqlClient; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +[Collection(SqlHistorianCollection.Name)] +public sealed class ProvisioningTests(SqlHistorianFixture fx) +{ + private (SqlServerTagProvisioner P, string S) Make(string? schema = null) + { + var s = schema ?? SqlHistorianFixture.UniqueSchema(); + var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s }; + var c = new SqlHistorianConnection(o); + return (new SqlServerTagProvisioner(c, new SqlServerSchemaBootstrapper(c, o)), s); + } + + private async Task<(byte Type, string? Eu, string? Desc)> TagAsync(string s, string name) + { + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = $"SELECT DataType, EngineeringUnit, Description FROM [{s}].Tag WHERE Name=@n"; + cmd.Parameters.AddWithValue("@n", name); + await using var r = await cmd.ExecuteReaderAsync(); + (await r.ReadAsync()).ShouldBeTrue(); + return (r.GetByte(0), r.IsDBNull(1) ? null : r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2)); + } + + [Fact] + public async Task Upsert_creates_then_changes_DataType_on_rerun_and_every_type_is_accepted() + { + fx.SkipIfUnavailable(); + var (p, s) = Make(); + var r1 = await p.EnsureTagsAsync([new("a", DriverDataType.Boolean, "degC", "desc"), new("s", DriverDataType.String, null, null)], CancellationToken.None); + r1.ShouldBe(new HistorianProvisionResult(2, 2, 0, 0)); + (await TagAsync(s, "a")).ShouldBe(((byte)DriverDataType.Boolean, "degC", "desc")); + var r2 = await p.EnsureTagsAsync([new("a", DriverDataType.Float64, null, null)], CancellationToken.None); + r2.ShouldBe(new HistorianProvisionResult(1, 1, 0, 0)); + (await TagAsync(s, "a")).ShouldBe(((byte)DriverDataType.Float64, "degC", "desc")); // null keeps stored metadata + } + + [Fact] + public async Task Duplicate_names_and_bad_names_are_tallied_not_thrown() + { + fx.SkipIfUnavailable(); + var (p, _) = Make(); + var r = await p.EnsureTagsAsync([ + new("d", DriverDataType.Int32, null, null), new("d", DriverDataType.Int64, null, null), + new("", DriverDataType.Int32, null, null), new(new string('x', 401), DriverDataType.Int32, null, null)], CancellationToken.None); + r.ShouldBe(new HistorianProvisionResult(4, 2, 0, 2)); + } + + [Fact] + public async Task More_tags_than_one_chunk_all_land() + { + fx.SkipIfUnavailable(); + var (p, s) = Make(); + var reqs = Enumerable.Range(0, 1000).Select(i => new HistorianTagProvisionRequest($"bulk{i}", DriverDataType.Float32, null, null)).ToList(); + (await p.EnsureTagsAsync(reqs, CancellationToken.None)).ShouldBe(new HistorianProvisionResult(1000, 1000, 0, 0)); + } + + [Fact] + public async Task Two_concurrent_provisioners_of_the_same_new_tags_both_report_Ensured_with_no_Failed() + { + fx.SkipIfUnavailable(); + var s = SqlHistorianFixture.UniqueSchema(); + var (p0, _) = Make(s); + await p0.EnsureTagsAsync([new("warm", DriverDataType.Int32, null, null)], CancellationToken.None); // bootstrap once + for (var round = 0; round < 10; round++) + { + var reqs = Enumerable.Range(0, 60).Select(i => new HistorianTagProvisionRequest($"r{round}.t{i}", DriverDataType.Float64, "u", null)).ToList(); + var rs = await Task.WhenAll(Enumerable.Range(0, 4).Select(_ => Task.Run(() => Make(s).P.EnsureTagsAsync(reqs, CancellationToken.None)))); + foreach (var r in rs) r.ShouldBe(new HistorianProvisionResult(60, 60, 0, 0)); + } + + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Tag"; + Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(601); + } + + [Fact] + public async Task A_connection_failure_is_a_Failed_tally_not_an_exception() + { + fx.SkipIfUnavailable(); + var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString + ";Connect Timeout=1", Schema = SqlHistorianFixture.UniqueSchema(), AutoCreateSchema = false }; + var c = new SqlHistorianConnection(o); + var p = new SqlServerTagProvisioner(c, new SqlServerSchemaBootstrapper(c, o)); + (await p.EnsureTagsAsync([new("x", DriverDataType.Int32, null, null)], CancellationToken.None)).ShouldBe(new HistorianProvisionResult(1, 0, 0, 1)); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs new file mode 100644 index 00000000..946ac321 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs @@ -0,0 +1,166 @@ +using System.Diagnostics; +using Microsoft.Data.SqlClient; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +[Collection(SqlHistorianCollection.Name)] +public sealed class WriteTests(SqlHistorianFixture fx) +{ + // Non-ms-aligned: sub-millisecond ticks must round-trip exactly. + private static readonly DateTime T0 = new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1234567); + + private (SqlServerHistorianValueWriter W, string S) Make() + { + var s = SqlHistorianFixture.UniqueSchema(); + var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s }; + var c = new SqlHistorianConnection(o); + var b = new SqlServerSchemaBootstrapper(c, o); + return (new SqlServerHistorianValueWriter(c, b, new SqlServerTagCatalog(c, b)), s); + } + + private async Task> RowsAsync(string schema, string tag) + { + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = $"SELECT s.SourceTimeUtc, s.ValueKind, s.NumValue, s.IntValue, s.StrValue, s.StatusCode, t.Name FROM [{schema}].Sample s JOIN [{schema}].Tag t ON t.TagId=s.TagId WHERE t.Name=@n ORDER BY s.SourceTimeUtc"; + cmd.Parameters.AddWithValue("@n", tag); + var l = new List<(DateTime, byte, double?, long?, string?, int, string)>(); + await using var r = await cmd.ExecuteReaderAsync(); + while (await r.ReadAsync()) + l.Add((SqlHistorianConnection.GetUtc(r, 0), r.GetByte(1), r.IsDBNull(2) ? null : r.GetDouble(2), r.IsDBNull(3) ? null : r.GetInt64(3), + r.IsDBNull(4) ? null : r.GetString(4), r.GetInt32(5), r.GetString(6))); + return l; + } + + [Fact] + public async Task Every_kind_round_trips_with_exact_ticks_and_status() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + var dt = new DateTime(2026, 1, 2, 3, 4, 5, DateTimeKind.Utc).AddTicks(77); + HistorizationValue[] vals = + [ + new(T0, HistorianValue.FromDouble(1.25), 0), + new(T0.AddTicks(1), HistorianValue.FromInt64(long.MaxValue), 0x40000000u), + new(T0.AddTicks(2), HistorianValue.FromBoolean(true), 0), + new(T0.AddTicks(3), HistorianValue.FromString("héllo"), 0), + new(T0.AddTicks(4), HistorianValue.FromDateTime(dt), 0), + new(T0.AddTicks(5), HistorianValue.Empty, 0x80310000u), + ]; + (await w.WriteLiveValuesAsync("tag.a", vals, CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(s, "tag.a"); + rows.Count.ShouldBe(6); + for (var i = 0; i < 6; i++) rows[i].T.Ticks.ShouldBe(vals[i].TimestampUtc!.Value.Ticks); + var back = rows.Select(r => SampleRowMapper.FromColumns(r.Kind, r.N, r.I, r.S)).ToList(); + back.ShouldBe(vals.Select(v => v.Value).ToList()); + rows[1].Status.ShouldBe(unchecked((int)0x40000000u)); + SampleRowMapper.FromDbStatus(rows[5].Status).ShouldBe(0x80310000u); + } + + [Fact] + public async Task Writing_the_same_batch_twice_is_idempotent() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + HistorizationValue[] vals = [new(T0, HistorianValue.FromDouble(1), 0), new(T0.AddSeconds(1), HistorianValue.FromDouble(2), 0)]; + (await w.WriteLiveValuesAsync("tag.b", vals, CancellationToken.None)).ShouldBeTrue(); + (await w.WriteLiveValuesAsync("tag.b", vals, CancellationToken.None)).ShouldBeTrue(); + // Overlapping batch: only the new row lands, the stored first value is untouched. + (await w.WriteLiveValuesAsync("tag.b", [new(T0, HistorianValue.FromDouble(99), 0), new(T0.AddSeconds(2), HistorianValue.FromDouble(3), 0)], CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(s, "tag.b"); + rows.Count.ShouldBe(3); + rows[0].N.ShouldBe(1); + } + + [Fact] + public async Task Within_batch_duplicate_timestamps_do_not_fail_and_first_wins() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + (await w.WriteLiveValuesAsync("tag.c", [new(T0, HistorianValue.FromDouble(1), 0), new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(s, "tag.c"); + rows.Count.ShouldBe(1); + rows[0].N.ShouldBe(1); + } + + [Fact] + public async Task Non_finite_doubles_store_as_Empty_with_their_status_and_do_not_fail_the_write() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + (await w.WriteLiveValuesAsync("tag.d", [ + new(T0, HistorianValue.FromDouble(double.NaN), 0x40000000u), + new(T0.AddTicks(1), HistorianValue.FromDouble(double.PositiveInfinity), 0), + new(T0.AddTicks(2), HistorianValue.FromDouble(7), 0)], CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(s, "tag.d"); + rows.Count.ShouldBe(3); + rows[0].Kind.ShouldBe((byte)HistorianValueKind.Empty); + rows[0].N.ShouldBeNull(); + rows[0].Status.ShouldBe(unchecked((int)0x40000000u)); + rows[2].N.ShouldBe(7); + } + + [Fact] + public async Task Empty_batch_succeeds_without_a_call_and_null_timestamp_is_stamped() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + (await w.WriteLiveValuesAsync("tag.e", [], CancellationToken.None)).ShouldBeTrue(); + (await w.WriteLiveValuesAsync("tag.e", [new(null, HistorianValue.FromInt64(5), 0)], CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(s, "tag.e"); + rows.Count.ShouldBe(1); + (DateTime.UtcNow - rows[0].T).ShouldBeLessThan(TimeSpan.FromMinutes(1)); + } + + [Fact] + public async Task Tag_is_created_with_the_first_typed_values_data_type() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + await w.WriteLiveValuesAsync("tag.f", [new(T0, HistorianValue.Empty, 0x80310000u), new(T0.AddTicks(1), HistorianValue.FromString("x"), 0)], CancellationToken.None); + await using var c = new SqlConnection(fx.ConnectionString); + await c.OpenAsync(); + await using var cmd = c.CreateCommand(); + cmd.CommandText = $"SELECT DataType FROM [{s}].Tag WHERE Name='tag.f'"; + Convert.ToByte(await cmd.ExecuteScalarAsync()).ShouldBe((byte)DriverDataType.String); + } + + [Fact] + public async Task Ten_thousand_values_write_in_under_five_seconds() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + await w.WriteLiveValuesAsync("tag.warm", [new(T0, HistorianValue.FromDouble(0), 0)], CancellationToken.None); // bootstrap + var vals = Enumerable.Range(0, 10_000).Select(i => new HistorizationValue(T0.AddMilliseconds(i), HistorianValue.FromDouble(i), 0)).ToArray(); + var sw = Stopwatch.StartNew(); + (await w.WriteLiveValuesAsync("tag.big", vals, CancellationToken.None)).ShouldBeTrue(); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5)); + (await RowsAsync(s, "tag.big")).Count.ShouldBe(10_000); + } + + [Fact] + public async Task Concurrent_overlapping_writers_all_succeed_and_leave_one_row_per_timestamp() + { + fx.SkipIfUnavailable(); + var (w, s) = Make(); + var vals = Enumerable.Range(0, 200).Select(i => new HistorizationValue(T0.AddMilliseconds(i), HistorianValue.FromDouble(i), 0)).ToArray(); + var results = await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => Task.Run(() => w.WriteLiveValuesAsync("tag.race", vals, CancellationToken.None)))); + results.ShouldAllBe(r => r); + (await RowsAsync(s, "tag.race")).Count.ShouldBe(200); + } + + [Fact] + public async Task A_SQL_failure_returns_false_instead_of_throwing() + { + fx.SkipIfUnavailable(); + var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString + ";Connect Timeout=1", Schema = SqlHistorianFixture.UniqueSchema(), AutoCreateSchema = false }; + var c = new SqlHistorianConnection(o); + var b = new SqlServerSchemaBootstrapper(c, o); + var w = new SqlServerHistorianValueWriter(c, b, new SqlServerTagCatalog(c, b)); + (await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeFalse(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs new file mode 100644 index 00000000..69d67dc0 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/AlarmEventIdTests.cs @@ -0,0 +1,47 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +public sealed class AlarmEventIdTests +{ + private static readonly DateTime T = new(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc); + + private static AlarmHistorianEvent Ev(string path = "P", string id = "A", string kind = "Activated", string user = "u", + string? comment = "c", string msg = "m", DateTime? t = null) => + new(id, path, "n", "LimitAlarm", AlarmSeverity.High, kind, msg, user, comment, t ?? T); + + [Fact] + public void Id_is_stable_lowercase_hex_sha256() + { + var a = SqlServerAlarmHistorianWriter.ComputeEventId(Ev()); + a.ShouldBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev())); + a.Length.ShouldBe(64); + a.ShouldBe(a.ToLowerInvariant()); + } + + [Fact] + public void Field_boundaries_are_unambiguous() + { + // The delimiter-based scheme collided on these; length prefixes must not. + SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "a|b", id: "c")) + .ShouldNotBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "a", id: "b|c"))); + SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "ab", id: "c")) + .ShouldNotBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "a", id: "bc"))); + } + + [Fact] + public void Null_and_empty_differ_and_every_field_participates() + { + var basis = SqlServerAlarmHistorianWriter.ComputeEventId(Ev()); + SqlServerAlarmHistorianWriter.ComputeEventId(Ev(comment: null)).ShouldNotBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev(comment: ""))); + foreach (var other in new[] + { + Ev(path: "Q"), Ev(id: "B"), Ev(kind: "Cleared"), Ev(user: "v"), Ev(comment: "d"), Ev(msg: "x"), + Ev(t: T.AddTicks(1)), + }) + SqlServerAlarmHistorianWriter.ComputeEventId(other).ShouldNotBe(basis); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ValueWriterPrepareTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ValueWriterPrepareTests.cs new file mode 100644 index 00000000..02b86fde --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/ValueWriterPrepareTests.cs @@ -0,0 +1,46 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests; + +public sealed class ValueWriterPrepareTests +{ + private static readonly DateTime T = new(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc); + + [Fact] + public void Non_finite_doubles_become_Empty_keeping_status() + { + const uint status = 0x40000000u; + var rows = SqlServerHistorianValueWriter.Prepare( + [ + new(T, HistorianValue.FromDouble(double.NaN), status), + new(T.AddTicks(1), HistorianValue.FromDouble(double.PositiveInfinity), 0), + new(T.AddTicks(2), HistorianValue.FromDouble(double.NegativeInfinity), 0), + new(T.AddTicks(3), HistorianValue.FromDouble(1.5), 0), + ], () => T); + rows.Select(r => r.Value.Kind).ShouldBe([HistorianValueKind.Empty, HistorianValueKind.Empty, HistorianValueKind.Empty, HistorianValueKind.Double]); + rows[0].StatusCode.ShouldBe(status); + } + + [Fact] + public void Duplicate_timestamps_keep_the_first_and_null_timestamp_uses_the_clock() + { + var rows = SqlServerHistorianValueWriter.Prepare( + [ + new(T, HistorianValue.FromInt64(1), 0), new(T, HistorianValue.FromInt64(2), 0), + new(null, HistorianValue.FromInt64(3), 0), + ], () => T.AddDays(1)); + rows.Count.ShouldBe(2); + rows[0].Value.Integer.ShouldBe(1); + rows[1].TimeUtc.ShouldBe(T.AddDays(1)); + } + + [Fact] + public void Data_type_mapping_is_one_table() + { + SqlHistorianDataTypes.FromKind(HistorianValueKind.Double).ShouldBe((byte)DriverDataType.Float64); + SqlHistorianDataTypes.FromKind(HistorianValueKind.Boolean).ShouldBe((byte)DriverDataType.Boolean); + SqlHistorianDataTypes.FromKind(HistorianValueKind.Empty).ShouldBe((byte)DriverDataType.Float64); + } +}