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

using Microsoft.Data.SqlClient;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;

/// <summary>
///     Seeds a fresh schema with plain INSERTs (independent of the value writer) and builds a data source over it.
///     Shared by the four read test classes. <see cref="T0"/> is deliberately NOT millisecond-aligned.
/// </summary>
internal sealed class ReadTestDb
{
    /// <summary>Base time: 2026-10-09 08:00:00.1234567 UTC (sub-millisecond ticks on purpose).</summary>
    public static readonly DateTime T0 = new DateTime(2026, 10, 9, 8, 0, 0, DateTimeKind.Utc).AddTicks(1_234_567);

    public const uint Good = 0x00000000u;
    public const uint Bad = 0x80000000u;
    public const uint BadNoCommunication = 0x80310000u;
    public const uint Uncertain = 0x40000000u;

    private ReadTestDb(SqlServerHistorianOptions options, SqlHistorianConnection conn, SqlServerSchemaBootstrapper boot)
    {
        Options = options;
        Conn = conn;
        Boot = boot;
    }

    public SqlServerHistorianOptions Options { get; }
    public SqlHistorianConnection Conn { get; }
    public SqlServerSchemaBootstrapper Boot { get; }

    public static DateTime At(double seconds) => T0.AddTicks((long)(seconds * TimeSpan.TicksPerSecond));

    public static async Task<ReadTestDb> CreateAsync(SqlHistorianFixture fx)
    {
        fx.SkipIfUnavailable();
        var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = SqlHistorianFixture.UniqueSchema() };
        var conn = new SqlHistorianConnection(o);
        var boot = new SqlServerSchemaBootstrapper(conn, o);
        await boot.EnsureAsync(CancellationToken.None);
        return new ReadTestDb(o, conn, boot);
    }

    /// <summary>A fresh data source (fresh catalog cache) over this schema; <paramref name="clock"/> is its "now".</summary>
    public SqlServerHistorianDataSource Source(TimeProvider? clock = null) =>
        new(Conn, Boot, new SqlServerTagCatalog(Conn, Boot), timeProvider: clock);

    /// <summary>Inserts a tag row and the given samples. ServerTimeUtc = source time + 1 s (so tests can tell them apart).</summary>
    public async Task<int> SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples)
    {
        await using var c = await Conn.OpenAsync(CancellationToken.None);
        int id;
        await using (var cmd = Conn.CreateCommand(c,
            $"INSERT {Conn.Q("Tag")} (Name, DataType, CreatedUtc, UpdatedUtc) OUTPUT inserted.TagId VALUES (@n, 0, SYSUTCDATETIME(), SYSUTCDATETIME())"))
        {
            cmd.Parameters.AddWithValue("@n", tag);
            id = (int)(await cmd.ExecuteScalarAsync())!;
        }
        foreach (var (t, v, status) in samples)
        {
            var col = SampleRowMapper.ToColumns(v);
            await using var cmd = Conn.CreateCommand(c,
                $"INSERT {Conn.Q("Sample")} (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " +
                "VALUES (@id, @t, @k, @num, @int, @str, @st, @srv)");
            cmd.Parameters.AddWithValue("@id", id);
            SqlHistorianConnection.AddDateTime2(cmd, "@t", t);
            cmd.Parameters.AddWithValue("@k", col.ValueKind);
            cmd.Parameters.Add(new SqlParameter("@num", System.Data.SqlDbType.Float) { Value = (object?)col.NumValue ?? DBNull.Value });
            cmd.Parameters.Add(new SqlParameter("@int", System.Data.SqlDbType.BigInt) { Value = (object?)col.IntValue ?? DBNull.Value });
            cmd.Parameters.Add(new SqlParameter("@str", System.Data.SqlDbType.NVarChar, -1) { Value = (object?)col.StrValue ?? DBNull.Value });
            cmd.Parameters.AddWithValue("@st", SampleRowMapper.ToDbStatus(status));
            SqlHistorianConnection.AddDateTime2(cmd, "@srv", t.AddSeconds(1));
            await cmd.ExecuteNonQueryAsync();
        }
        return id;
    }

    /// <summary>The fixture tag A of the brief: +0=10, +10=20, +20=Bad(30), +30=40, +40=50 (Double).</summary>
    public Task<int> SeedAAsync(string name = "A") => SeedAsync(name,
        (At(0), HistorianValue.FromDouble(10), Good),
        (At(10), HistorianValue.FromDouble(20), Good),
        (At(20), HistorianValue.FromDouble(30), Bad),
        (At(30), HistorianValue.FromDouble(40), Good),
        (At(40), HistorianValue.FromDouble(50), Good));

    public async Task<long> ScalarAsync(string sql)
    {
        await using var c = await Conn.OpenAsync(CancellationToken.None);
        await using var cmd = Conn.CreateCommand(c, sql);
        return Convert.ToInt64(await cmd.ExecuteScalarAsync());
    }
}

[Collection(SqlHistorianCollection.Name)]
public sealed class RawReadTests(SqlHistorianFixture fx)
{
    private static readonly CancellationToken Ct = CancellationToken.None;
    private static DateTime At(double s) => ReadTestDb.At(s);

    private static double?[] Values(HistoryReadResult r) => r.Samples.Select(s => (double?)s.Value).ToArray();

    [Fact]
    public async Task Forward_read_is_start_inclusive_end_exclusive_ascending_with_exact_ticks()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAAsync();
        var r = await db.Source().ReadRawAsync("A", At(10), At(40), 0, Ct);

        Values(r).ShouldBe([20d, 30d, 40d]); // +10 included, +40 (== end) excluded
        r.Samples.Select(s => s.SourceTimestampUtc!.Value.Ticks).ShouldBe([At(10).Ticks, At(20).Ticks, At(30).Ticks]);
        r.Samples.Select(s => s.StatusCode).ShouldBe([ReadTestDb.Good, ReadTestDb.Bad, ReadTestDb.Good]);
        r.Samples[0].SourceTimestampUtc!.Value.Kind.ShouldBe(DateTimeKind.Utc);
        r.Samples[0].ServerTimestampUtc.Ticks.ShouldBe(At(11).Ticks); // stored ServerTimeUtc, not "now"
        r.ContinuationPoint.ShouldBeNull();
    }

    [Fact]
    public async Task Forward_read_honours_the_limit()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAAsync();
        Values(await db.Source().ReadRawAsync("A", At(0), At(100), 2, Ct)).ShouldBe([10d, 20d]);
    }

    [Fact]
    public async Task Zero_max_values_is_unbounded()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAAsync();
        Values(await db.Source().ReadRawAsync("A", At(-1), At(100), 0, Ct)).ShouldBe([10d, 20d, 30d, 40d, 50d]);
        Values(await db.Source().ReadRawAsync("A", At(-1), At(100), uint.MaxValue, Ct)).Length.ShouldBe(5);
    }

    [Fact]
    public async Task Reverse_read_is_descending_over_end_exclusive_start_inclusive()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAAsync();
        // start=+30, end=+10 ⇒ (+10, +30] descending.
        Values(await db.Source().ReadRawAsync("A", At(30), At(10), 0, Ct)).ShouldBe([40d, 30d]);
        Values(await db.Source().ReadRawAsync("A", At(100), At(-1), 2, Ct)).ShouldBe([50d, 40d]);
    }

    [Fact]
    public async Task Equal_start_and_end_is_a_point_read()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAAsync();
        Values(await db.Source().ReadRawAsync("A", At(10), At(10), 0, Ct)).ShouldBe([20d]);
        (await db.Source().ReadRawAsync("A", At(10).AddTicks(1), At(10).AddTicks(1), 0, Ct)).Samples.ShouldBeEmpty();
    }

    [Fact]
    public async Task A_resumed_page_from_the_last_timestamp_inclusive_continues_without_gaps()
    {
        // The node manager's paging: read cap N, resume at the last returned timestamp INCLUSIVE, trim it.
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAAsync();
        var src = db.Source();
        var p1 = await src.ReadRawAsync("A", At(0), At(100), 2, Ct);
        var p2 = await src.ReadRawAsync("A", p1.Samples[^1].SourceTimestampUtc!.Value, At(100), 2, Ct);
        Values(p1).ShouldBe([10d, 20d]);
        Values(p2).ShouldBe([20d, 30d]); // first is the boundary sample the node manager trims
    }

    [Fact]
    public async Task Unknown_tag_is_an_empty_result_and_reads_never_create_tags()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        var src = db.Source();
        (await src.ReadRawAsync("nope", At(0), At(10), 0, Ct)).Samples.ShouldBeEmpty();
        await src.ReadAtTimeAsync("nope", [At(1)], Ct);
        await src.ReadProcessedAsync("nope", At(0), At(10), TimeSpan.FromSeconds(5), HistoryAggregateType.Average, Ct);
        (await db.ScalarAsync($"SELECT COUNT(*) FROM {db.Conn.Q("Tag")}")).ShouldBe(0);
        src.GetHealthSnapshot().TotalFailures.ShouldBe(0);
    }

    [Fact]
    public async Task Outage_markers_and_bad_samples_come_back_with_their_bad_status()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAsync("M",
            (At(0), HistorianValue.FromDouble(10), ReadTestDb.Good),
            (At(5), HistorianValue.Empty, ReadTestDb.BadNoCommunication),
            (At(7), HistorianValue.FromDouble(99), ReadTestDb.Bad));
        var r = await db.Source().ReadRawAsync("M", At(0), At(10), 0, Ct);
        r.Samples.Count.ShouldBe(3);
        r.Samples[1].Value.ShouldBeNull();
        r.Samples[1].StatusCode.ShouldBe(0x80310000u);
        r.Samples[2].Value.ShouldBe(99d);
        r.Samples[2].StatusCode.ShouldBe(0x80000000u);
    }

    [Fact]
    public async Task Every_kind_round_trips_to_its_CLR_type()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        var dt = new DateTime(2025, 1, 2, 3, 4, 5, DateTimeKind.Utc).AddTicks(7);
        await db.SeedAsync("K",
            (At(0), HistorianValue.FromInt64(-9_000_000_000L), ReadTestDb.Good),
            (At(1), HistorianValue.FromBoolean(true), ReadTestDb.Good),
            (At(2), HistorianValue.FromString("héllo"), ReadTestDb.Good),
            (At(3), HistorianValue.FromDateTime(dt), ReadTestDb.Good),
            (At(4), HistorianValue.FromDouble(1.5), ReadTestDb.Uncertain));
        var s = (await db.Source().ReadRawAsync("K", At(0), At(5), 0, Ct)).Samples;
        s[0].Value.ShouldBe(-9_000_000_000L);
        s[1].Value.ShouldBe(true);
        s[2].Value.ShouldBe("héllo");
        s[3].Value.ShouldBe(dt);
        ((DateTime)s[3].Value!).Kind.ShouldBe(DateTimeKind.Utc);
        s[4].Value.ShouldBe(1.5);
        s[4].StatusCode.ShouldBe(0x40000000u);
    }

    [Fact]
    public async Task Successful_reads_are_counted_in_health()
    {
        var db = await ReadTestDb.CreateAsync(fx);
        await db.SeedAAsync();
        var src = db.Source();
        await src.ReadRawAsync("A", At(0), At(10), 0, Ct);
        await src.ReadRawAsync("A", At(0), At(10), 0, Ct);
        var h = src.GetHealthSnapshot();
        h.TotalQueries.ShouldBe(2);
        h.TotalSuccesses.ShouldBe(2);
        h.ConsecutiveFailures.ShouldBe(0);
        h.LastSuccessTime.ShouldNotBeNull();
        h.ProcessConnectionOpen.ShouldBeTrue();
    }

    [Fact]
    public async Task A_missing_schema_surfaces_in_health_LastError_and_is_not_latched()
    {
        fx.SkipIfUnavailable();
        var o = new SqlServerHistorianOptions
        {
            ConnectionString = fx.ConnectionString, Schema = SqlHistorianFixture.UniqueSchema(), AutoCreateSchema = false,
        };
        var conn = new SqlHistorianConnection(o);
        var clock = new FixedClock(DateTimeOffset.UtcNow);
        var boot = new SqlServerSchemaBootstrapper(conn, o, timeProvider: clock);
        var src = new SqlServerHistorianDataSource(conn, boot, new SqlServerTagCatalog(conn, boot));

        await Should.ThrowAsync<HistorianSchemaMissingException>(() => src.ReadRawAsync("A", At(0), At(1), 0, Ct));
        await Should.ThrowAsync<HistorianSchemaMissingException>(() => src.ReadEventsAsync(null, At(0), At(1), 0, Ct));
        var h = src.GetHealthSnapshot();
        h.TotalFailures.ShouldBe(2);
        h.ConsecutiveFailures.ShouldBe(2);
        h.LastError!.ShouldContain("missing version 1");
        h.ProcessConnectionOpen.ShouldBeFalse();

        // Another node creates the schema; the same data source then succeeds (bootstrap failure is not latched).
        var creator = new SqlServerHistorianOptions { ConnectionString = o.ConnectionString, Schema = o.Schema };
        await new SqlServerSchemaBootstrapper(new SqlHistorianConnection(creator), creator).EnsureAsync(Ct);
        clock.Now += SqlServerSchemaBootstrapper.FailureCacheDuration; // past the 5 s failure cache (M-2)
        (await src.ReadRawAsync("A", At(0), At(1), 0, Ct)).Samples.ShouldBeEmpty();
        src.GetHealthSnapshot().ConsecutiveFailures.ShouldBe(0);
    }
}