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

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<List<(DateTime T, byte Kind, double? N, long? I, string? S, int Status, string Tag)>> 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[5].N, rows[5].I, rows[5].S).ShouldBe((null, null, null));
        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();
        // A deadlock victim (1205) legitimately returns false and the recorder retries; model that, and require convergence.
        var results = await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => Task.Run(async () =>
        {
            for (var i = 0; i < 5; i++)
                if (await w.WriteLiveValuesAsync("tag.race", vals, CancellationToken.None)) return true;
            return false;
        })));
        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();
    }
}