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; /// QuestDB-specific checks of (the literal codes run through conformance in T7). [Collection(QuestDbCollection.Name)] public sealed class UpdateTests(QuestDbFixture fx) { private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); // µs-aligned private const uint Good = HistorianStatusCodes.Good; internal sealed record Rig(QuestDbConnection Conn, QuestDbHistorianDataUpdater Updater, QuestDbTagCatalog Tags); internal static Rig MakeRig(QuestDbFixture fx, TimeSpan? visibility = null) { var o = fx.Options(fx.UniquePrefix()); o.VisibilityTimeout = visibility ?? o.VisibilityTimeout; var c = new QuestDbConnection(o); var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout); var tags = new QuestDbTagCatalog(c, new QuestDbSchemaBootstrapper(c, o), wal); return new Rig(c, new QuestDbHistorianDataUpdater(c, tags, wal), tags); } /// The HistoryUpdate caller these tests pass to the updater (recorded in the modification trail). internal static readonly HistoryUpdateContext Who = new("questdb-test", null); internal static HistorizationValue V(DateTime? t, double v, uint status = Good) => new(t, HistorianValue.FromDouble(v), status); internal sealed record Row(DateTime Ts, byte Kind, double? Num, long? Status, bool Deleted); /// Every stored row for the tag (tombstones included), straight from the table. internal static async Task> RowsAsync(Rig rig, string tag, bool liveOnly = false) { await using var conn = await rig.Conn.OpenAsync(default); await using var cmd = rig.Conn.CreateCommand(conn, $"SELECT ts, kind, num_value, status, deleted FROM {rig.Conn.SampleTable} WHERE tag = $1 {(liveOnly ? "AND NOT deleted" : "")} ORDER BY ts"); QuestDbConnection.BindText(cmd, tag); var rows = new List(); await using var r = await cmd.ExecuteReaderAsync(); while (await r.ReadAsync()) rows.Add(new Row(QuestDbConnection.ReadUtc(r, 0), QuestDbConnection.ReadKind(r, 1), r.IsDBNull(2) ? null : r.GetDouble(2), r.IsDBNull(3) ? null : r.GetInt64(3), r.GetBoolean(4))); return rows; } internal static async Task TagRowsAsync(Rig rig, string tag) { await using var conn = await rig.Conn.OpenAsync(default); await using var cmd = rig.Conn.CreateCommand(conn, $"SELECT count() FROM {rig.Conn.TagTable} WHERE name = $1"); QuestDbConnection.BindText(cmd, tag); return Convert.ToInt64(await cmd.ExecuteScalarAsync()); } [Fact] public async Task Insert_writes_absent_and_refuses_existing_and_is_visible_immediately() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); (await rig.Updater.UpdateDataAsync("t1", HistoryUpdateMode.Insert, [V(T0, 1)], UpdateTests.Who, default)) .ShouldBe([HistorianStatusCodes.GoodEntryInserted]); var codes = await rig.Updater.UpdateDataAsync("t1", HistoryUpdateMode.Insert, [V(T0, 99), V(T0.AddSeconds(1), 2)], UpdateTests.Who, default); codes.ShouldBe([HistorianStatusCodes.BadEntryExists, HistorianStatusCodes.GoodEntryInserted]); var rows = await RowsAsync(rig, "t1"); // read right after return: read-your-writes rows.Select(r => r.Num).ShouldBe([1d, 2d]); (await TagRowsAsync(rig, "t1")).ShouldBe(1); } [Fact] public async Task Replace_overwrites_present_refuses_missing_and_never_creates_the_tag() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); (await rig.Updater.UpdateDataAsync("nope", HistoryUpdateMode.Replace, [V(T0, 1)], UpdateTests.Who, default)) .ShouldBe([HistorianStatusCodes.BadNoEntryExists]); (await TagRowsAsync(rig, "nope")).ShouldBe(0); (await RowsAsync(rig, "nope")).ShouldBeEmpty(); await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1)], UpdateTests.Who, default); var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [V(T0, 5), V(T0.AddSeconds(1), 6)], UpdateTests.Who, default); codes.ShouldBe([HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.BadNoEntryExists]); (await RowsAsync(rig, "t")).Select(r => r.Num).ShouldBe([5d]); } [Fact] public async Task Update_upserts_both_ways_and_last_duplicate_wins() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1)], UpdateTests.Who, default); var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Update, [V(T0, 2), V(T0.AddSeconds(1), 3), V(T0.AddSeconds(1).AddTicks(3), 4)], UpdateTests.Who, default); codes.ShouldBe([HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.GoodEntryInserted, HistorianStatusCodes.GoodEntryInserted]); (await RowsAsync(rig, "t")).Select(r => r.Num).ShouldBe([2d, 4d]); } [Fact] public async Task Same_microsecond_duplicates_collapse_before_the_database_sees_them() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); // offsets 0 and 3 ticks sit inside one 10-tick window => the same stored µs var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1), V(T0.AddTicks(3), 2)], UpdateTests.Who, default); codes.ShouldBe([HistorianStatusCodes.GoodEntryInserted, HistorianStatusCodes.BadEntryExists]); (await RowsAsync(rig, "t")).Select(r => r.Num).ShouldBe([1d]); } [Fact] public async Task Invalid_items_get_the_planners_codes_and_the_rest_proceed() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); var codes = await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [ V(null, 1), V(DateTime.MinValue, 1), V(T0, double.NaN), new HistorizationValue(T0.AddSeconds(1), HistorianValue.Empty, Good), V(T0.AddSeconds(2), 7), ], UpdateTests.Who, default); codes.ShouldBe([HistorianStatusCodes.BadInvalidTimestamp, HistorianStatusCodes.BadInvalidTimestamp, HistorianStatusCodes.BadInvalidArgument, HistorianStatusCodes.BadInvalidArgument, HistorianStatusCodes.GoodEntryInserted]); (await RowsAsync(rig, "t")).Count.ShouldBe(1); } [Fact] public async Task Over_long_tag_is_refused_and_nothing_is_written() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); var tag = new string('x', HistorianTagNames.MaxLength + 1); (await rig.Updater.UpdateDataAsync(tag, HistoryUpdateMode.Insert, [V(T0, 1)], UpdateTests.Who, default)) .ShouldBe([HistorianStatusCodes.BadInvalidArgument]); (await rig.Updater.DeleteAtTimeAsync(tag, [T0], UpdateTests.Who, default)).ShouldBe([HistorianStatusCodes.BadInvalidArgument]); (await rig.Updater.DeleteRawAsync(tag, T0, T0.AddHours(1), UpdateTests.Who, default)).ShouldBe(HistorianStatusCodes.BadInvalidArgument); } [Fact] public async Task Pre_1970_items_are_BadOutOfRange_and_nothing_is_sent() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); var pre = new DateTime(1969, 12, 31, 23, 59, 59, DateTimeKind.Utc); foreach (var mode in new[] { HistoryUpdateMode.Insert, HistoryUpdateMode.Replace, HistoryUpdateMode.Update }) { var codes = await rig.Updater.UpdateDataAsync("t", mode, [V(pre, 1), V(DateTime.MinValue, 1), V(T0, 2)], UpdateTests.Who, default); codes[0].ShouldBe(HistorianStatusCodes.BadOutOfRange); codes[1].ShouldBe(HistorianStatusCodes.BadInvalidTimestamp); } // Pre-1970 only: the whole batch is refused, no tag row is created, no INSERT would succeed anyway. (await rig.Updater.UpdateDataAsync("only-pre", HistoryUpdateMode.Update, [V(pre, 1)], UpdateTests.Who, default)) .ShouldBe([HistorianStatusCodes.BadOutOfRange]); (await TagRowsAsync(rig, "only-pre")).ShouldBe(0); (await rig.Updater.DeleteAtTimeAsync("t", [pre], UpdateTests.Who, default)).ShouldBe([HistorianStatusCodes.BadNoEntryExists]); } [Fact] public async Task DeleteAtTime_tombstones_live_rows_and_reports_per_time_codes() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1), V(T0.AddSeconds(1), 2)], UpdateTests.Who, default); var codes = await rig.Updater.DeleteAtTimeAsync("t", [T0, T0.AddSeconds(5), T0, DateTime.MinValue], UpdateTests.Who, default); codes.ShouldBe([Good, HistorianStatusCodes.BadNoEntryExists, HistorianStatusCodes.BadNoEntryExists, HistorianStatusCodes.BadInvalidTimestamp]); (await RowsAsync(rig, "t", liveOnly: true)).Select(r => r.Num).ShouldBe([2d]); // a tombstone is not a live row: deleting it again finds nothing (await rig.Updater.DeleteAtTimeAsync("t", [T0], UpdateTests.Who, default)).ShouldBe([HistorianStatusCodes.BadNoEntryExists]); (await rig.Updater.DeleteAtTimeAsync("unknown", [T0], UpdateTests.Who, default)).ShouldBe([HistorianStatusCodes.BadNoEntryExists]); } [Fact] public async Task DeleteAtTime_with_a_sub_microsecond_request_time_still_reports_Good_for_the_row_it_deleted() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1)], UpdateTests.Who, default); // +7 ticks is inside the same µs; the row lives at T0. (await rig.Updater.DeleteAtTimeAsync("t", [T0.AddTicks(7)], UpdateTests.Who, default)).ShouldBe([Good]); (await RowsAsync(rig, "t", liveOnly: true)).ShouldBeEmpty(); } [Fact] public async Task DeleteRaw_is_half_open_pages_through_big_ranges_and_reports_BadNoData_when_empty() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); (await rig.Updater.DeleteRawAsync("unknown", T0, T0.AddHours(1), UpdateTests.Who, default)).ShouldBe(HistorianStatusCodes.BadNoData); // 2,500 rows => 3 pages of at most 1,000; plus one row exactly at the (exclusive) end. var values = Enumerable.Range(0, 2500).Select(i => V(T0.AddSeconds(i), i)).ToList(); values.Add(V(T0.AddSeconds(2500), -1)); (await rig.Updater.UpdateDataAsync("big", HistoryUpdateMode.Insert, values, UpdateTests.Who, default)) .ShouldAllBe(c => c == HistorianStatusCodes.GoodEntryInserted); (await rig.Updater.DeleteRawAsync("big", T0, T0.AddSeconds(2500), UpdateTests.Who, default)).ShouldBe(Good); var live = await RowsAsync(rig, "big", liveOnly: true); live.Count.ShouldBe(1); live[0].Num.ShouldBe(-1d); (await rig.Updater.DeleteRawAsync("big", T0, T0.AddSeconds(2500), UpdateTests.Who, default)).ShouldBe(HistorianStatusCodes.BadNoData); } [Fact] public async Task DeleteRaw_with_a_pre_1970_start_reads_from_the_bound_unclamped() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(T0, 1), V(T0.AddSeconds(1), 2)], UpdateTests.Who, default); (await rig.Updater.DeleteRawAsync("t", new DateTime(1900, 1, 1, 0, 0, 0, DateTimeKind.Utc), T0.AddSeconds(1), UpdateTests.Who, default)) .ShouldBe(Good); (await RowsAsync(rig, "t", liveOnly: true)).Select(r => r.Num).ShouldBe([2d]); // DateTime.MinValue as the start is nudged inside, not bound as -infinity (await rig.Updater.DeleteRawAsync("t", DateTime.MinValue, DateTime.MaxValue, UpdateTests.Who, default)).ShouldBe(Good); (await RowsAsync(rig, "t", liveOnly: true)).ShouldBeEmpty(); } [Fact] public async Task Read_after_write_is_visible_across_many_immediate_calls() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); // ~15% of un-waited inserts are not yet visible (spike Q4), so 40 round trips expose a missing wait. for (var i = 0; i < 40; i++) { var t = T0.AddSeconds(i); (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [V(t, i)], UpdateTests.Who, default)) .ShouldBe([HistorianStatusCodes.GoodEntryInserted]); (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [V(t, i + 1000)], UpdateTests.Who, default)) .ShouldBe([HistorianStatusCodes.GoodEntryReplaced], $"iteration {i}: the insert was not visible"); } } [Fact] public async Task Concurrent_inserts_of_the_same_new_key_yield_exactly_one_GoodEntryInserted() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); for (var i = 0; i < 20; i++) // loop so the race shows reliably without the lock { var t = T0.AddSeconds(i); var codes = await Task.WhenAll( Task.Run(() => rig.Updater.UpdateDataAsync("race", HistoryUpdateMode.Insert, [V(t, 1)], UpdateTests.Who, default)), Task.Run(() => rig.Updater.UpdateDataAsync("race", HistoryUpdateMode.Insert, [V(t, 2)], UpdateTests.Who, default))); codes.Select(c => c[0]).OrderBy(c => c).ToArray().ShouldBe( new[] { HistorianStatusCodes.GoodEntryInserted, HistorianStatusCodes.BadEntryExists }, $"iteration {i}"); } } [Fact] public async Task Existence_reads_are_bounded_when_request_times_are_far_apart() { fx.SkipIfUnavailable(); var rig = MakeRig(fx); // 1 Hz seed across 2 h = 7,200 live rows; the request touches 3 instants 1 h apart. var seed = Enumerable.Range(0, 7200).Select(i => V(T0.AddSeconds(i), i)).ToList(); await rig.Updater.UpdateDataAsync("sparse", HistoryUpdateMode.Insert, seed, UpdateTests.Who, default); var before = rig.Updater.LiveRowsRead; var codes = await rig.Updater.UpdateDataAsync("sparse", HistoryUpdateMode.Replace, [V(T0, 1), V(T0.AddHours(1), 2), V(T0.AddHours(2).AddSeconds(-1), 3), V(T0.AddHours(1).AddMilliseconds(500), 4)], UpdateTests.Who, default); codes.ShouldBe([HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.GoodEntryReplaced, HistorianStatusCodes.BadNoEntryExists]); // 3 clusters of a single instant each (T0+1h and +1h.5s are one cluster of 2 instants, 1 live row) (rig.Updater.LiveRowsRead - before).ShouldBeLessThanOrEqualTo(3); } }