diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAlarmHistorianWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAlarmHistorianWriter.cs new file mode 100644 index 00000000..9b39a5d6 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAlarmHistorianWriter.cs @@ -0,0 +1,130 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Npgsql; +using NpgsqlTypes; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; + +/// +/// over the QuestDB event table. Each event carries the deterministic +/// and is inserted through the table's DEDUP UPSERT KEYS(ts, event_id) +/// (T0 Q3 confirmed VARCHAR works as a key), so the store-and-forward drain's at-least-once redelivery, and both +/// pair nodes writing the same event, converge on one row (the redelivered row is identical except +/// received_utc). Outcomes: Ack on insert; RetryPlease on transient / unclassified errors and on +/// cancellation; PermanentFail only for data the server will always reject (a time before 1970, which is not a +/// storable designated timestamp, or a server message naming an inconvertible / out-of-range value). Never throws. +/// +internal sealed class QuestDbAlarmHistorianWriter : IAlarmHistorianWriter +{ + private static readonly string[] PermanentMessageFragments = + ["designated timestamp", "inconvertible", "out of range", "too long", "cannot be null"]; + + private readonly QuestDbConnection _conn; + private readonly QuestDbSchemaBootstrapper _bootstrap; + private readonly QuestDbWalVisibility _wal; + private readonly ILogger _log; + private readonly TimeProvider _clock; + + /// Creates the writer. + /// Connection helper. + /// Schema bootstrapper. + /// WAL visibility wait (once per batch, so a following read sees the events). + /// Optional logger. + /// Clock for received_utc. + public QuestDbAlarmHistorianWriter(QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbWalVisibility wal, + ILogger? log = null, TimeProvider? clock = null) + { + _conn = conn; + _bootstrap = bootstrap; + _wal = wal; + _log = log ?? NullLogger.Instance; + _clock = clock ?? TimeProvider.System; + } + + /// + 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("QuestDB alarm historian bootstrap failed ({Exception}); will retry.", ex.GetType().Name); + Array.Fill(outcomes, HistorianWriteOutcome.RetryPlease); + return outcomes; + } + + var wrote = false; + 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); + wrote |= outcomes[i] == HistorianWriteOutcome.Ack; + } + + if (wrote) + { + try { await _wal.WaitAsync(_conn.EventTable, cancellationToken).ConfigureAwait(false); } + catch (Exception ex) { _log.LogDebug("QuestDB alarm visibility wait failed ({Exception}).", ex.GetType().Name); } + } + return outcomes; + } + + private async Task WriteOneAsync(AlarmHistorianEvent e, CancellationToken ct) + { + try + { + if (!QuestDbConnection.IsStorable(e.TimestampUtc)) + { + _log.LogWarning("QuestDB alarm event dropped: its time is before 1970-01-01 and cannot be stored."); + return HistorianWriteOutcome.PermanentFail; + } + + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, + $"INSERT INTO {_conn.EventTable} (ts, event_id, source, alarm_id, alarm_name, alarm_type_name, severity, event_kind, message, user_name, comment, received_utc) " + + "VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)"); + QuestDbConnection.BindTimestamp(cmd, e.TimestampUtc); + QuestDbConnection.BindText(cmd, AlarmEventIds.ComputeEventId(e)); + QuestDbConnection.BindText(cmd, e.EquipmentPath ?? ""); + QuestDbConnection.BindText(cmd, e.AlarmId ?? ""); + QuestDbConnection.BindText(cmd, e.AlarmName ?? ""); + QuestDbConnection.BindText(cmd, e.AlarmTypeName ?? ""); + cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = AlarmEventIds.ToNumericSeverity(e.Severity) }); + QuestDbConnection.BindText(cmd, e.EventKind ?? ""); + QuestDbConnection.BindText(cmd, e.Message); + QuestDbConnection.BindText(cmd, e.User); + QuestDbConnection.BindText(cmd, e.Comment); + QuestDbConnection.BindTimestamp(cmd, _clock.GetUtcNow().UtcDateTime); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + return HistorianWriteOutcome.Ack; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return HistorianWriteOutcome.RetryPlease; + } + catch (Exception ex) + { + var outcome = Classify(ex); + _log.LogWarning("QuestDB alarm event write failed ({Exception}); outcome {Outcome}.", ex.GetType().Name, outcome); + return outcome; + } + } + + // Only a data error QuestDB will reject forever dead-letters; anything unknown or non-database retries (never lose alarm + // history). SQLSTATE is always 00000 on QuestDB, so a server rejection is classified by message text. + internal static HistorianWriteOutcome Classify(Exception ex) + { + if (QuestDbConnection.IsTransient(ex)) return HistorianWriteOutcome.RetryPlease; + if (ex is PostgresException pg && + PermanentMessageFragments.Any(f => pg.MessageText.Contains(f, StringComparison.OrdinalIgnoreCase))) + return HistorianWriteOutcome.PermanentFail; + return HistorianWriteOutcome.RetryPlease; + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs new file mode 100644 index 00000000..791ef010 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianValueWriter.cs @@ -0,0 +1,180 @@ +using System.Text; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Npgsql; +using NpgsqlTypes; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; + +/// +/// over the QuestDB sample table. Redelivery-idempotent with first-wins +/// semantics: QuestDB's DEDUP UPSERT KEYS(ts, tag) would let a re-sent sample overwrite the stored one (and a +/// re-sent live sample resurrect a tombstone), so the writer reads which timestamps already exist for the tag in +/// the batch's range (tombstones count as present, T0 Q11 range form) and inserts only the absent rows. +/// Never throws; returns false on any failure so the recorder keeps its entries. +/// +/// +/// Batch normalisation is at µs resolution (non-finite doubles become Empty +/// + Bad, an in-batch µs collision keeps the first). RC1: a sample whose time is before 1970 cannot be a +/// QuestDB designated timestamp and one such row would fail the whole multi-row INSERT, so those samples are dropped +/// (Warning once per batch, counted in ) and the call still succeeds for the rest. +/// R5: after inserting, the writer waits for WAL visibility so an immediate redelivery sees the rows. +/// Inserts are chunked at rows (8 bound parameters per row, far under the 32,767 limit). +/// +internal sealed class QuestDbHistorianValueWriter : IHistorianValueWriter +{ + /// Rows per multi-row INSERT (T0 Q2: 1,000 rows is reliable). + internal const int BatchRows = 1000; + + private readonly QuestDbConnection _conn; + private readonly QuestDbTagCatalog _catalog; + private readonly QuestDbWalVisibility _wal; + private readonly ILogger _log; + private readonly Func _utcNow; + private long _droppedPre1970; + + /// Creates the writer. + /// Connection helper. + /// Tag catalog (the only tag authority). + /// WAL visibility wait. + /// Optional logger. + /// Clock for null timestamps and server_ts (tests). + public QuestDbHistorianValueWriter(QuestDbConnection conn, QuestDbTagCatalog catalog, QuestDbWalVisibility wal, + ILogger? log = null, Func? utcNow = null) + { + _conn = conn; + _catalog = catalog; + _wal = wal; + _log = log ?? NullLogger.Instance; + _utcNow = utcNow ?? (() => DateTime.UtcNow); + } + + /// Samples dropped so far because their time is before 1970 (RC1). + internal long DroppedPre1970 => Interlocked.Read(ref _droppedPre1970); + + /// + public async Task WriteLiveValuesAsync(string tag, IReadOnlyList values, CancellationToken ct) + { + try + { + if (!HistorianTagNames.IsValid(tag)) + { + _log.LogWarning("QuestDB historian live-value write refused: tag name is missing or longer than {Max} characters.", + HistorianTagNames.MaxLength); + return false; + } + if (values is null || values.Count == 0) return true; + + var prepared = LiveWritePreparer.Prepare(values, _utcNow, resolutionTicks: 10); + var rows = prepared.Where(r => QuestDbConnection.IsStorable(r.TimeUtc)).ToList(); + var dropped = prepared.Count - rows.Count; + if (dropped > 0) + { + Interlocked.Add(ref _droppedPre1970, dropped); + _log.LogWarning("QuestDB historian dropped {Count} sample(s) for a tag because their time is before 1970-01-01 (not storable).", dropped); + } + if (rows.Count == 0) return true; + + var dataType = HistorianDataTypes.FromKind(rows.Select(r => r.Value.Kind).FirstOrDefault(k => k != HistorianValueKind.Empty)); + await _catalog.EnsureAsync(tag, dataType, ct).ConfigureAwait(false); + + var present = await ExistingTicksAsync(tag, rows, ct).ConfigureAwait(false); + var fresh = rows.Where(r => !present.Contains(QuestDbConnection.Round(r.TimeUtc).Ticks)).ToList(); + if (fresh.Count == 0) return true; + + var serverTs = _utcNow(); + for (var i = 0; i < fresh.Count; i += BatchRows) + await InsertAsync(tag, fresh.GetRange(i, Math.Min(BatchRows, fresh.Count - i)), serverTs, ct).ConfigureAwait(false); + + await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); // timeout/suspension does not change the result (R5) + return true; + } + catch (Exception ex) + { + _log.LogWarning("QuestDB historian live-value write failed ({Exception}); will retry.", ex.GetType().Name); + return false; + } + } + + /// A gap between consecutive batch times larger than this starts a new existence-query range. + internal static readonly TimeSpan ExistenceClusterGap = TimeSpan.FromMinutes(1); + + /// + /// Splits ascending into inclusive [Lo, Hi] ranges: a new range starts where the gap to the + /// previous time exceeds or the range already holds times. One outlier + /// time therefore costs its own tiny query instead of widening one range over every stored row in between. + /// + /// Times in ascending order. + /// Largest gap kept inside one range. + /// Most times in one range (>= 1). + /// The ranges, ascending. + internal static IReadOnlyList<(DateTime Lo, DateTime Hi)> Cluster(IReadOnlyList sortedTimes, TimeSpan gap, int maxPerCluster) + { + ArgumentOutOfRangeException.ThrowIfLessThan(maxPerCluster, 1); + var result = new List<(DateTime, DateTime)>(); + if (sortedTimes.Count == 0) return result; + var lo = sortedTimes[0]; + var prev = lo; + var n = 1; + for (var i = 1; i < sortedTimes.Count; i++) + { + var t = sortedTimes[i]; + if (t - prev > gap || n >= maxPerCluster) + { + result.Add((lo, prev)); + lo = t; + n = 0; + } + prev = t; + n++; + } + result.Add((lo, prev)); + return result; + } + + // T0 Q11 range form, one query per cluster: tombstones are NOT filtered out (a deleted row must not be resurrected). + private async Task> ExistingTicksAsync(string tag, List rows, CancellationToken ct) + { + var times = rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).OrderBy(t => t).ToList(); + var set = new HashSet(); + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + foreach (var (lo, hi) in Cluster(times, ExistenceClusterGap, BatchRows)) + { + await using var cmd = _conn.CreateCommand(conn, $"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3"); + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindTimestamp(cmd, lo); + QuestDbConnection.BindTimestamp(cmd, hi); + await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await r.ReadAsync(ct).ConfigureAwait(false)) set.Add(QuestDbConnection.ReadUtc(r, 0).Ticks); + } + return set; + } + + private async Task InsertAsync(string tag, List chunk, DateTime serverTs, CancellationToken ct) + { + var sql = new StringBuilder($"INSERT INTO {_conn.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES "); + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, ""); + var n = 1; + for (var i = 0; i < chunk.Count; i++) + { + if (i > 0) sql.Append(','); + sql.Append($"(${n},${n + 1},${n + 2},${n + 3},${n + 4},${n + 5},${n + 6},false,${n + 7})"); + n += 8; + var row = chunk[i]; + var c = SampleColumnMapper.ToColumns(row.Value); + QuestDbConnection.BindTimestamp(cmd, row.TimeUtc); + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindByte(cmd, c.ValueKind); + cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Double, Value = c.NumValue is { } d ? d : DBNull.Value }); + cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Bigint, Value = c.IntValue is { } l ? l : DBNull.Value }); + QuestDbConnection.BindText(cmd, c.StrValue); + QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(row.StatusCode)); + QuestDbConnection.BindTimestamp(cmd, serverTs); + } + cmd.CommandText = sql.ToString(); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagProvisioner.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagProvisioner.cs new file mode 100644 index 00000000..f32f9828 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagProvisioner.cs @@ -0,0 +1,141 @@ +using System.Text; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; + +/// +/// for the QuestDB backend: upserts tag rows (fixed designated timestamp +/// 1970-01-01, so the dedup key makes it an upsert by name). QuestDB replaces the WHOLE row on a key hit, so a null +/// engineering unit / description in a request would wipe the stored one; per chunk the existing rows are read +/// (WHERE name IN (…)) and COALESCE(request, stored) is applied client-side before inserting. +/// Every data type is provisionable (Skipped is always 0). Never throws. After each chunk it waits for WAL +/// visibility and tells the catalog the names are known (R7). +/// +internal sealed class QuestDbTagProvisioner : IHistorianProvisioning +{ + private const int ChunkSize = 400; // 6 parameters per tag on insert, 400 on the IN list + private const int MaxUnitLength = 64; + private const int MaxDescriptionLength = 400; + + private readonly QuestDbConnection _conn; + private readonly QuestDbSchemaBootstrapper _bootstrap; + private readonly QuestDbTagCatalog _catalog; + private readonly QuestDbWalVisibility _wal; + private readonly ILogger _log; + private readonly TimeProvider _clock; + + /// Creates the provisioner. + /// Connection helper. + /// Schema bootstrapper. + /// Tag catalog (told which names are now known). + /// WAL visibility wait. + /// Optional logger. + /// Clock for updated_utc. + public QuestDbTagProvisioner(QuestDbConnection conn, QuestDbSchemaBootstrapper bootstrap, QuestDbTagCatalog catalog, + QuestDbWalVisibility wal, ILogger? log = null, TimeProvider? clock = null) + { + _conn = conn; + _bootstrap = bootstrap; + _catalog = catalog; + _wal = wal; + _log = log ?? NullLogger.Instance; + _clock = clock ?? TimeProvider.System; + } + + /// + 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 (last request wins); over-long/empty names fail alone. Weight = requests folded into the name. + var byName = new Dictionary(StringComparer.Ordinal); + foreach (var r in requests!) + { + if (r is null || string.IsNullOrWhiteSpace(r.TagName) || !HistorianTagNames.IsValid(r.TagName)) { 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 TryUpsertAsync(chunk.Select(c => c.Req).ToList(), ct).ConfigureAwait(false)) ensured += weight; + else failed += weight; + } + } + catch (Exception ex) + { + _log.LogWarning("QuestDB historian provisioning failed ({Exception}).", ex.GetType().Name); + failed = requested - ensured; + } + + return new HistorianProvisionResult(requested, ensured, 0, failed); + } + + private async Task TryUpsertAsync(List chunk, CancellationToken ct) + { + try + { + var stored = new Dictionary(StringComparer.Ordinal); + await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false)) + { + var sel = new StringBuilder($"SELECT name, eng_unit, description FROM {_conn.TagTable} WHERE name IN ("); + await using var cmd = _conn.CreateCommand(conn, ""); + for (var i = 0; i < chunk.Count; i++) + { + sel.Append(i == 0 ? "$" : ",$").Append(i + 1); + QuestDbConnection.BindText(cmd, chunk[i].TagName); + } + cmd.CommandText = sel.Append(')').ToString(); + await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await r.ReadAsync(ct).ConfigureAwait(false)) + stored[r.GetString(0)] = (r.IsDBNull(1) ? null : r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2)); + } + + var now = _clock.GetUtcNow().UtcDateTime; + await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false)) + { + var ins = new StringBuilder($"INSERT INTO {_conn.TagTable} (ts, name, data_type, eng_unit, description, updated_utc) VALUES "); + await using var cmd = _conn.CreateCommand(conn, ""); + var n = 1; + for (var i = 0; i < chunk.Count; i++) + { + var req = chunk[i]; + stored.TryGetValue(req.TagName, out var old); + if (i > 0) ins.Append(','); + ins.Append($"(${n},${n + 1},${n + 2},${n + 3},${n + 4},${n + 5})"); + n += 6; + QuestDbConnection.BindTimestamp(cmd, QuestDbTables.TagEpoch); + QuestDbConnection.BindText(cmd, req.TagName); + QuestDbConnection.BindByte(cmd, (byte)req.DataType); + QuestDbConnection.BindText(cmd, Clip(req.EngineeringUnit ?? old.Unit, MaxUnitLength)); + QuestDbConnection.BindText(cmd, Clip(req.Description ?? old.Desc, MaxDescriptionLength)); + QuestDbConnection.BindTimestamp(cmd, now); + } + cmd.CommandText = ins.ToString(); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } + + await _wal.WaitAsync(_conn.TagTable, ct).ConfigureAwait(false); + foreach (var c in chunk) _catalog.MarkKnown(c.TagName); + return true; + } + catch (Exception ex) + { + _log.LogWarning("QuestDB historian tag upsert failed ({Exception}).", ex.GetType().Name); + return false; + } + } + + private static string? Clip(string? s, int max) => s is null ? null : s.Length <= max ? s : s[..max]; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs new file mode 100644 index 00000000..032ffa86 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs @@ -0,0 +1,118 @@ +using Npgsql; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests; + +[Collection(QuestDbCollection.Name)] +public sealed class AlarmWriterTests(QuestDbFixture fx) +{ + // µs-aligned: QuestDB stores µs, so the ts column reads back exactly. + private static readonly DateTime T = new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1230); + + private (QuestDbAlarmHistorianWriter W, QuestDbConnection C) Make(string? connection = null, bool auto = true) + { + var o = fx.Options(fx.UniquePrefix(), autoCreate: auto); + if (connection is not null) o.ConnectionString = connection; + var c = new QuestDbConnection(o); + return (new QuestDbAlarmHistorianWriter(c, new QuestDbSchemaBootstrapper(c, o), new QuestDbWalVisibility(c, o.VisibilityTimeout)), c); + } + + private static AlarmHistorianEvent Ev(string kind = "Activated", string type = "LimitAlarm", string? comment = "c", DateTime? at = null) => + new("alarm-1", "Plant/Line/Eq", "HiTemp", type, AlarmSeverity.High, kind, "msg", "user", comment, at ?? T); + + private static async Task CountAsync(QuestDbConnection c, string? kind = null) + { + await using var conn = await c.OpenAsync(default); + await using var cmd = c.CreateCommand(conn, $"SELECT count() FROM {c.EventTable}" + (kind is null ? "" : " WHERE event_kind = $1")); + if (kind is not null) QuestDbConnection.BindText(cmd, kind); + return Convert.ToInt64(await cmd.ExecuteScalarAsync()); + } + + [Fact] + public async Task Event_round_trips_and_a_repeat_is_one_row_with_every_write_Ack() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + (await w.WriteBatchAsync([Ev(), Ev()], CancellationToken.None)).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 conn = await c.OpenAsync(default); + await using var cmd = c.CreateCommand(conn, + $"SELECT event_id, source, alarm_id, alarm_name, alarm_type_name, severity, message, user_name, comment, ts, received_utc FROM {c.EventTable} WHERE event_kind = 'Activated'"); + await using var r = await cmd.ExecuteReaderAsync(); + (await r.ReadAsync()).ShouldBeTrue(); + r.GetString(0).ShouldBe(AlarmEventIds.ComputeEventId(Ev())); + r.GetString(1).ShouldBe("Plant/Line/Eq"); + r.GetString(2).ShouldBe("alarm-1"); + r.GetString(3).ShouldBe("HiTemp"); + r.GetString(4).ShouldBe("LimitAlarm"); + r.GetInt32(5).ShouldBe(AlarmEventIds.ToNumericSeverity(AlarmSeverity.High)); + r.GetInt32(5).ShouldBe(700); + r.GetString(6).ShouldBe("msg"); + r.GetString(7).ShouldBe("user"); + r.GetString(8).ShouldBe("c"); + QuestDbConnection.ReadUtc(r, 9).ShouldBe(T); + (DateTime.UtcNow - QuestDbConnection.ReadUtc(r, 10)).ShouldBeLessThan(TimeSpan.FromMinutes(1)); + (await r.ReadAsync()).ShouldBeFalse(); + await r.CloseAsync(); + (await CountAsync(c)).ShouldBe(2); + } + + [Fact] + public async Task A_null_comment_and_a_very_long_type_name_are_both_stored_and_Ack() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", type: new string('t', 500)), Ev(kind: "C", comment: null)], CancellationToken.None); + r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]); + (await CountAsync(c)).ShouldBe(3); + } + + [Fact] + public async Task An_event_before_1970_is_PermanentFail_and_siblings_still_Ack() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", at: new DateTime(1960, 1, 1, 0, 0, 0, DateTimeKind.Utc)), Ev(kind: "C")], CancellationToken.None); + r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack]); + (await CountAsync(c)).ShouldBe(2); + } + + [Fact] + public async Task Concurrent_writers_of_the_same_event_all_Ack_with_one_row() + { + fx.SkipIfUnavailable(); + var (w, c) = 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 CountAsync(c, "Activated")).ShouldBe(1); + } + + [Fact] + public async Task Unreachable_database_is_RetryPlease_not_an_exception_and_empty_batch_is_empty() + { + fx.SkipIfUnavailable(); + var (w, _) = Make("Host=localhost;Port=1;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading;Timeout=2", auto: false); + (await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.RetryPlease]); + (await w.WriteBatchAsync([], CancellationToken.None)).ShouldBeEmpty(); + } + + [Fact] + public void Classify_maps_unknown_and_transport_errors_to_RetryPlease_and_data_rejections_to_PermanentFail() + { + QuestDbAlarmHistorianWriter.Classify(new InvalidOperationException("boom")).ShouldBe(HistorianWriteOutcome.RetryPlease); + QuestDbAlarmHistorianWriter.Classify(new IOException("reset")).ShouldBe(HistorianWriteOutcome.RetryPlease); + QuestDbAlarmHistorianWriter.Classify(new TimeoutException()).ShouldBe(HistorianWriteOutcome.RetryPlease); + QuestDbAlarmHistorianWriter.Classify(new PostgresException("table is busy", "ERROR", "ERROR", "00000")).ShouldBe(HistorianWriteOutcome.RetryPlease); + QuestDbAlarmHistorianWriter.Classify(new PostgresException("designated timestamp before 1970-01-01 is not allowed", "ERROR", "ERROR", "00000")) + .ShouldBe(HistorianWriteOutcome.PermanentFail); + QuestDbAlarmHistorianWriter.Classify(new PostgresException("inconvertible value: x [VARCHAR -> INT]", "ERROR", "ERROR", "00000")) + .ShouldBe(HistorianWriteOutcome.PermanentFail); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ProvisioningTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ProvisioningTests.cs new file mode 100644 index 00000000..e1b8605e --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ProvisioningTests.cs @@ -0,0 +1,102 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests; + +[Collection(QuestDbCollection.Name)] +public sealed class ProvisioningTests(QuestDbFixture fx) +{ + private (QuestDbTagProvisioner P, QuestDbConnection C, QuestDbTagCatalog Cat) Make(string? prefix = null, string? connectionString = null, + bool autoCreate = true) + { + var o = fx.Options(prefix ?? fx.UniquePrefix(), autoCreate: autoCreate); + if (connectionString is not null) o.ConnectionString = connectionString; + var c = new QuestDbConnection(o); + var b = new QuestDbSchemaBootstrapper(c, o); + var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout); + var cat = new QuestDbTagCatalog(c, b, wal); + return (new QuestDbTagProvisioner(c, b, cat, wal), c, cat); + } + + private static async Task<(byte Type, string? Eu, string? Desc)> TagAsync(QuestDbConnection c, string name) + { + await using var conn = await c.OpenAsync(default); + await using var cmd = c.CreateCommand(conn, $"SELECT data_type, eng_unit, description FROM {c.TagTable} WHERE name = $1"); + QuestDbConnection.BindText(cmd, name); + await using var r = await cmd.ExecuteReaderAsync(); + (await r.ReadAsync()).ShouldBeTrue(); + return (QuestDbConnection.ReadKind(r, 0), r.IsDBNull(1) ? null : r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2)); + } + + private static async Task CountAsync(QuestDbConnection c) + { + await using var conn = await c.OpenAsync(default); + await using var cmd = c.CreateCommand(conn, $"SELECT count() FROM {c.TagTable}"); + return Convert.ToInt64(await cmd.ExecuteScalarAsync()); + } + + [Fact] + public async Task Upsert_creates_then_changes_DataType_on_rerun_and_every_type_is_accepted() + { + fx.SkipIfUnavailable(); + var (p, c, _) = 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(c, "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(c, "a")).ShouldBe(((byte)DriverDataType.Float64, "degC", "desc")); // null keeps stored metadata + (await CountAsync(c)).ShouldBe(2); // upsert by name, not a second row + } + + [Fact] + public async Task A_request_with_values_replaces_only_the_fields_it_carries() + { + fx.SkipIfUnavailable(); + var (p, c, _) = Make(); + await p.EnsureTagsAsync([new("m", DriverDataType.Int32, "A", "first")], CancellationToken.None); + await p.EnsureTagsAsync([new("m", DriverDataType.Int32, "B", null)], CancellationToken.None); + (await TagAsync(c, "m")).ShouldBe(((byte)DriverDataType.Int32, "B", "first")); + } + + [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, c, _) = 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)); + (await CountAsync(c)).ShouldBe(1000); + } + + [Fact] + public async Task A_later_catalog_Ensure_of_a_provisioned_tag_does_not_wipe_its_metadata() + { + fx.SkipIfUnavailable(); + var (p, c, cat) = Make(); + await p.EnsureTagsAsync([new("k", DriverDataType.Float64, "kg", "mass")], CancellationToken.None); + (await cat.ExistsAsync("k", CancellationToken.None)).ShouldBeTrue(); + await cat.EnsureAsync("k", 1, CancellationToken.None); + (await TagAsync(c, "k")).ShouldBe(((byte)DriverDataType.Float64, "kg", "mass")); + } + + [Fact] + public async Task A_connection_failure_is_a_Failed_tally_not_an_exception() + { + fx.SkipIfUnavailable(); + var (p, _, _) = Make(connectionString: "Host=localhost;Port=1;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading;Timeout=2"); + (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.QuestDb.IntegrationTests/WriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs new file mode 100644 index 00000000..39e75e1b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs @@ -0,0 +1,296 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests; + +[Collection(QuestDbCollection.Name)] +public sealed class WriteTests(QuestDbFixture fx) +{ + // µs-aligned (ticks % 10 == 0) so "same µs" offsets of a few ticks stay inside one 10-tick window. + private static readonly DateTime T0 = new(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc); + private static readonly DateTime ServerNow = new(2026, 10, 9, 12, 0, 0, DateTimeKind.Utc); + + private (QuestDbHistorianValueWriter W, QuestDbConnection C) Make(string? prefix = null, string? connectionString = null) + { + var o = fx.Options(prefix ?? fx.UniquePrefix()); + if (connectionString is not null) o.ConnectionString = connectionString; + var c = new QuestDbConnection(o); + var b = new QuestDbSchemaBootstrapper(c, o); + var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout); + return (new QuestDbHistorianValueWriter(c, new QuestDbTagCatalog(c, b, wal), wal, utcNow: () => ServerNow), c); + } + + private static async Task> RowsAsync( + QuestDbConnection c, string tag) + { + await using var conn = await c.OpenAsync(default); + await using var cmd = c.CreateCommand(conn, + $"SELECT ts, kind, num_value, int_value, str_value, status, deleted, server_ts FROM {c.SampleTable} WHERE tag = $1 ORDER BY ts"); + QuestDbConnection.BindText(cmd, tag); + var l = new List<(DateTime, byte, double?, long?, string?, long, bool, DateTime)>(); + await using var r = await cmd.ExecuteReaderAsync(); + while (await r.ReadAsync()) + l.Add((QuestDbConnection.ReadUtc(r, 0), QuestDbConnection.ReadKind(r, 1), r.IsDBNull(2) ? null : r.GetDouble(2), + r.IsDBNull(3) ? null : r.GetInt64(3), r.IsDBNull(4) ? null : r.GetString(4), r.GetInt64(5), r.GetBoolean(6), + QuestDbConnection.ReadUtc(r, 7))); + return l; + } + + private static async Task SeedTombstoneAsync(QuestDbConnection c, QuestDbWalVisibility wal, string tag, DateTime t) + { + await using (var conn = await c.OpenAsync(default)) + { + await using var cmd = c.CreateCommand(conn, + $"INSERT INTO {c.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ($1,$2,$3,NULL,NULL,NULL,$4,true,$5)"); + QuestDbConnection.BindTimestamp(cmd, t); + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindByte(cmd, 0); + QuestDbConnection.BindLong(cmd, 0); + QuestDbConnection.BindTimestamp(cmd, t); + await cmd.ExecuteNonQueryAsync(); + } + await wal.WaitAsync(c.SampleTable, default); + } + + [Fact] + public async Task All_six_kinds_round_trip_with_a_non_Good_status() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + const uint uncertain = HistorianStatusCodes.UncertainDataSubNormal; + const uint bad = HistorianStatusCodes.BadOutOfRange; + var dt = new DateTime(2025, 5, 6, 7, 8, 9, DateTimeKind.Utc).AddTicks(1230); + HistorizationValue[] vals = + [ + new(T0, HistorianValue.FromDouble(1.5), uncertain), + new(T0.AddSeconds(1), HistorianValue.FromInt64(-77), uncertain), + new(T0.AddSeconds(2), HistorianValue.FromBoolean(true), uncertain), + new(T0.AddSeconds(3), HistorianValue.FromString("héllo"), uncertain), + new(T0.AddSeconds(4), HistorianValue.FromDateTime(dt), uncertain), + new(T0.AddSeconds(5), HistorianValue.Empty, bad), + ]; + (await w.WriteLiveValuesAsync("tag.kinds", vals, CancellationToken.None)).ShouldBeTrue(); + + var rows = await RowsAsync(c, "tag.kinds"); + rows.Count.ShouldBe(6); + for (var i = 0; i < 6; i++) + { + rows[i].T.ShouldBe(vals[i].TimestampUtc!.Value); + rows[i].Status.ShouldBe(SampleColumnMapper.ToLongStatus(vals[i].StatusCode)); + rows[i].Deleted.ShouldBeFalse(); + rows[i].ServerTs.ShouldBe(ServerNow); + SampleColumnMapper.FromColumns(rows[i].Kind, rows[i].N, rows[i].I, rows[i].S).ShouldBe(vals[i].Value); + } + } + + [Fact] + public async Task A_redelivered_batch_leaves_one_row_and_the_first_value() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + (await w.WriteLiveValuesAsync("tag.redeliver", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue(); + (await w.WriteLiveValuesAsync("tag.redeliver", [new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(c, "tag.redeliver"); + rows.Count.ShouldBe(1); + rows[0].N.ShouldBe(1d); + } + + [Fact] + public async Task Redelivery_straight_after_the_write_always_sees_it_because_the_writer_waits_for_WAL_visibility() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + // Without the visibility wait ~35 % of immediate re-reads miss the row (T0 Q4) and the second value would replace the first. + for (var i = 0; i < 40; i++) + { + var t = T0.AddSeconds(i); + (await w.WriteLiveValuesAsync("tag.rw", [new(t, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue(); + (await w.WriteLiveValuesAsync("tag.rw", [new(t, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue(); + } + var rows = await RowsAsync(c, "tag.rw"); + rows.Count.ShouldBe(40); + rows.ShouldAllBe(r => r.N == 1d); + } + + [Fact] + public async Task A_tombstoned_key_is_not_resurrected_by_redelivery() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5)); + (await w.WriteLiveValuesAsync("tag.tomb", [new(T0, HistorianValue.FromDouble(9), 0)], CancellationToken.None)).ShouldBeTrue(); + await SeedTombstoneAsync(c, wal, "tag.tomb", T0.AddSeconds(1)); + + // Redeliver the tombstoned key plus a fresh neighbour. + (await w.WriteLiveValuesAsync("tag.tomb", + [new(T0.AddSeconds(1), HistorianValue.FromDouble(5), 0), new(T0.AddSeconds(2), HistorianValue.FromDouble(6), 0)], + CancellationToken.None)).ShouldBeTrue(); + + var rows = await RowsAsync(c, "tag.tomb"); + rows.Count.ShouldBe(3); + rows.Single(r => r.T == T0.AddSeconds(1)).Deleted.ShouldBeTrue(); + rows.Single(r => r.T == T0.AddSeconds(2)).N.ShouldBe(6d); + } + + [Fact] + public async Task NaN_is_stored_as_Empty_with_BadOutOfRange() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + (await w.WriteLiveValuesAsync("tag.nan", [new(T0, HistorianValue.FromDouble(double.NaN), 0)], CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(c, "tag.nan"); + rows.Count.ShouldBe(1); + rows[0].Kind.ShouldBe((byte)HistorianValueKind.Empty); + rows[0].Status.ShouldBe((long)HistorianStatusCodes.BadOutOfRange); + rows[0].N.ShouldBeNull(); + } + + [Fact] + public async Task A_microsecond_collision_within_a_batch_keeps_the_first() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + // +2 and +7 ticks stay inside one 10-tick (1 µs) window of the aligned T0. + (await w.WriteLiveValuesAsync("tag.us", + [new(T0.AddTicks(2), HistorianValue.FromDouble(1), 0), new(T0.AddTicks(7), HistorianValue.FromDouble(2), 0)], + CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(c, "tag.us"); + rows.Count.ShouldBe(1); + rows[0].N.ShouldBe(1d); + rows[0].T.ShouldBe(T0); + } + + [Fact] + public async Task Samples_before_1970_are_dropped_and_the_rest_still_land() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var old = new DateTime(1960, 1, 1, 0, 0, 0, DateTimeKind.Utc); + (await w.WriteLiveValuesAsync("tag.old", + [new(old, HistorianValue.FromDouble(1), 0), new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue(); + w.DroppedPre1970.ShouldBe(1); + var rows = await RowsAsync(c, "tag.old"); + rows.Count.ShouldBe(1); + rows[0].T.ShouldBe(T0); + + // A batch of nothing but unstorable samples succeeds (nothing to retry) and stores nothing. + (await w.WriteLiveValuesAsync("tag.old2", [new(old, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue(); + w.DroppedPre1970.ShouldBe(2); + (await RowsAsync(c, "tag.old2")).ShouldBeEmpty(); + } + + [Fact] + public async Task A_batch_larger_than_one_insert_chunk_lands_completely() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var vals = Enumerable.Range(0, 2500).Select(i => new HistorizationValue(T0.AddMilliseconds(i), HistorianValue.FromInt64(i), 0)).ToArray(); + (await w.WriteLiveValuesAsync("tag.big", vals, CancellationToken.None)).ShouldBeTrue(); + var rows = await RowsAsync(c, "tag.big"); + rows.Count.ShouldBe(2500); + rows[1999].I.ShouldBe(1999); + } + + [Fact] + public async Task An_over_long_tag_name_returns_false() + { + fx.SkipIfUnavailable(); + var (w, _) = Make(); + (await w.WriteLiveValuesAsync(new string('x', 401), [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeFalse(); + } + + [Fact] + public async Task A_stopped_port_returns_false_without_throwing() + { + fx.SkipIfUnavailable(); + var (w, _) = Make(connectionString: "Host=localhost;Port=1;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading;Timeout=2"); + (await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeFalse(); + } + + [Fact] + public async Task A_cancelled_write_returns_false_without_throwing() + { + fx.SkipIfUnavailable(); + var (w, _) = Make(); + using var cts = new CancellationTokenSource(); + cts.Cancel(); + (await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], cts.Token)).ShouldBeFalse(); + } + + private static async Task SeedRowAsync(QuestDbConnection c, QuestDbWalVisibility wal, string tag, DateTime t, double v) + { + await using (var conn = await c.OpenAsync(default)) + { + await using var cmd = c.CreateCommand(conn, + $"INSERT INTO {c.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ($1,$2,1,$3,NULL,NULL,0,false,$4)"); + QuestDbConnection.BindTimestamp(cmd, t); + QuestDbConnection.BindText(cmd, tag); + cmd.Parameters.Add(new Npgsql.NpgsqlParameter { NpgsqlDbType = NpgsqlTypes.NpgsqlDbType.Double, Value = v }); + QuestDbConnection.BindTimestamp(cmd, t); + await cmd.ExecuteNonQueryAsync(); + } + await wal.WaitAsync(c.SampleTable, default); + } + + [Fact] + public async Task A_row_on_another_tag_at_the_same_timestamp_does_not_suppress_the_write() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5)); + (await w.WriteLiveValuesAsync("tag.A", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue(); + (await w.WriteLiveValuesAsync("tag.B", [new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue(); + (await RowsAsync(c, "tag.A")).Single().N.ShouldBe(1d); + (await RowsAsync(c, "tag.B")).Single().N.ShouldBe(2d); + } + + [Fact] + public async Task An_absent_interior_timestamp_is_written_while_both_present_ends_are_left_alone() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + var wal = new QuestDbWalVisibility(c, TimeSpan.FromSeconds(5)); + await w.WriteLiveValuesAsync("tag.ends", [new(T0.AddSeconds(100), HistorianValue.FromDouble(0), 0)], CancellationToken.None); // creates the tag/tables + await SeedRowAsync(c, wal, "tag.ends", T0, 10); + await SeedRowAsync(c, wal, "tag.ends", T0.AddSeconds(20), 30); + (await w.WriteLiveValuesAsync("tag.ends", + [new(T0, HistorianValue.FromDouble(-1), 0), new(T0.AddSeconds(10), HistorianValue.FromDouble(20), 0), new(T0.AddSeconds(20), HistorianValue.FromDouble(-3), 0)], + CancellationToken.None)).ShouldBeTrue(); + var rows = (await RowsAsync(c, "tag.ends")).Where(r => r.T <= T0.AddSeconds(20)).ToList(); + rows.Select(r => r.N).ShouldBe([10d, 20d, 30d]); + } + + [Fact] + public async Task An_outlier_timestamp_does_not_widen_the_existence_read_over_the_stored_rows_between() + { + fx.SkipIfUnavailable(); + var (w, c) = Make(); + // Behavioural check of the whole path with a far-apart batch: both samples land, a redelivery adds nothing. + var far = T0.AddYears(1); + HistorizationValue[] vals = [new(T0, HistorianValue.FromDouble(1), 0), new(far, HistorianValue.FromDouble(2), 0)]; + (await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue(); + (await w.WriteLiveValuesAsync("tag.far", vals, CancellationToken.None)).ShouldBeTrue(); + (await RowsAsync(c, "tag.far")).Count.ShouldBe(2); + } + + [Fact] + public void Cluster_splits_on_a_gap_over_the_limit_and_at_the_per_cluster_cap() + { + var g = TimeSpan.FromMinutes(1); + QuestDbHistorianValueWriter.Cluster([], g, 10).ShouldBeEmpty(); + QuestDbHistorianValueWriter.Cluster([T0], g, 10).ShouldBe([(T0, T0)]); + // exactly the gap stays together; one tick over splits + QuestDbHistorianValueWriter.Cluster([T0, T0 + g], g, 10).ShouldBe([(T0, T0 + g)]); + QuestDbHistorianValueWriter.Cluster([T0, T0 + g + TimeSpan.FromTicks(10)], g, 10) + .ShouldBe([(T0, T0), (T0 + g + TimeSpan.FromTicks(10), T0 + g + TimeSpan.FromTicks(10))]); + // outlier gets its own range + var far = T0.AddYears(1); + QuestDbHistorianValueWriter.Cluster([T0, T0.AddSeconds(1), far], g, 10).ShouldBe([(T0, T0.AddSeconds(1)), (far, far)]); + // cap: 5 close times, max 2 per cluster + var five = Enumerable.Range(0, 5).Select(i => T0.AddSeconds(i)).ToList(); + QuestDbHistorianValueWriter.Cluster(five, g, 2) + .ShouldBe([(five[0], five[1]), (five[2], five[3]), (five[4], five[4])]); + } +}