lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/WriteTests.cs

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<List<(DateTime T, byte Kind, double? N, long? I, string? S, long Status, bool Deleted, DateTime ServerTs)>> 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_drain_pass_waits_for_wal_visibility_once_not_once_per_tag_group()
    {
        fx.SkipIfUnavailable();
        var o = fx.Options(fx.UniquePrefix());
        var c = new QuestDbConnection(o);
        var b = new QuestDbSchemaBootstrapper(c, o);
        var catalogWal = new QuestDbWalVisibility(c, o.VisibilityTimeout);
        var sampleWal = new QuestDbWalVisibility(c, o.VisibilityTimeout); // the writer's own: counts only sample waits
        var w = new QuestDbHistorianValueWriter(c, new QuestDbTagCatalog(c, b, catalogWal), sampleWal, utcNow: () => ServerNow);

        foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" })
            (await w.WriteLiveValuesInPassAsync(tag, [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeTrue();
        sampleWal.Waits.ShouldBe(0, "in-pass writes defer the visibility wait");

        await w.FlushAsync(CancellationToken.None);
        sampleWal.Waits.ShouldBe(1, "one wait for the whole pass");
        foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" })
            (await RowsAsync(c, tag)).Count.ShouldBe(1, $"{tag} is visible after the flush");

        // A redelivery of the pass (the insert-if-absent check reads what the flush made visible) keeps one row each —
        // and, having inserted nothing, owes no wait.
        foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" })
            (await w.WriteLiveValuesInPassAsync(tag, [new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue();
        await w.FlushAsync(CancellationToken.None);
        foreach (var tag in new[] { "pass.a", "pass.b", "pass.c" })
            (await RowsAsync(c, tag)).Single().N.ShouldBe(1d);
        sampleWal.Waits.ShouldBe(1);

        // The plain per-call path still waits per call (positive control).
        (await w.WriteLiveValuesAsync("pass.d", [new(T0, HistorianValue.FromDouble(4), 0)], CancellationToken.None)).ShouldBeTrue();
        sampleWal.Waits.ShouldBe(2);
    }

    [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_rethrows_so_the_recorder_does_not_book_a_tag_failure()
    {
        // The recorder's contract: a cancelled drain pass is not this tag's failure (it used to return false, which the
        // recorder counted toward the tag's quarantine budget).
        fx.SkipIfUnavailable();
        var (w, _) = Make();
        using var cts = new CancellationTokenSource();
        cts.Cancel();
        await Should.ThrowAsync<OperationCanceledException>(
            () => w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], cts.Token));
    }

    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);
    }
}