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;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

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 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_write_cancelled_mid_statement_rethrows_instead_of_failing_the_tag()
    {
        // lmxopcua#778: SqlClient reports a command cancelled mid-flight as a SqlException, which the writer booked as a
        // failed write (false) — a shutdown could tip a healthy tag over its attempt budget. It must be a cancellation.
        fx.SkipIfUnavailable();
        var (w, s) = Make();
        (await w.WriteLiveValuesAsync("tag.cancel", [new(T0, HistorianValue.FromDouble(0), 0)], CancellationToken.None))
            .ShouldBeTrue("bootstrap + tag id");

        await using var blocker = new SqlConnection(fx.ConnectionString);
        var testCt = TestContext.Current.CancellationToken;
        await blocker.OpenAsync(testCt);
        await using var tx = (SqlTransaction)await blocker.BeginTransactionAsync(testCt);
        await using (var hold = new SqlCommand($"SELECT COUNT(*) FROM [{s}].Sample WITH (TABLOCKX, HOLDLOCK)", blocker, tx))
            await hold.ExecuteScalarAsync(testCt);

        using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(500));
        var ex = await Should.ThrowAsync<OperationCanceledException>(() =>
            w.WriteLiveValuesAsync("tag.cancel", [new(T0.AddSeconds(1), HistorianValue.FromDouble(1), 0)], cts.Token));
        ex.CancellationToken.ShouldBe(cts.Token);
        await tx.RollbackAsync(testCt);

        // Positive control: with the lock gone the same write succeeds.
        (await w.WriteLiveValuesAsync("tag.cancel", [new(T0.AddSeconds(1), HistorianValue.FromDouble(1), 0)], CancellationToken.None))
            .ShouldBeTrue();
    }

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