using System.Data;
using System.Data.Common;
using System.Diagnostics;
using Microsoft.Data.Sqlite;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;

namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests;

/// <summary>
///     Proves <see cref="SqlPollReader"/> against a real database (<see cref="SqlitePollFixture"/>): the
///     N-in/N-out ordering contract, the slice-back of one result set to many tags, the three distinct
///     no-value outcomes (absent row / NULL cell / unresolvable ref), and — the part that cannot be proven
///     by reading the code — that a query which refuses to return <b>does not wedge the caller</b>.
///     <para><b>Each test owns its own fixture.</b> The fixture is a temp file, so a fresh one per test is
///     cheap and buys total isolation — which matters here because two tests deliberately mutate the
///     database (an extra duplicate row; an EXCLUSIVE transaction) in ways a shared fixture would leak into
///     every other test in the class.</para>
/// </summary>
public sealed class SqlPollReaderTests
{
    private static readonly TimeSpan CommandTimeout = TimeSpan.FromSeconds(10);
    private static readonly TimeSpan OperationTimeout = TimeSpan.FromSeconds(15);

    // ---- the plan's headline contract: N in, N out, in input order ----

    [Fact]
    public async Task ReadAsync_returnsSnapshotsInInputOrder_absentKeyIsBadNoData()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture,
            KvTag("Speed", SqlitePollFixture.PresentKey),
            KvTag("Missing", SqlitePollFixture.AbsentKey),
            KvTag("Temp", SqlitePollFixture.NullValueKey));

        var snapshots = await reader.ReadAsync(["Speed", "Missing", "Temp"], CancellationToken.None);

        snapshots.Count.ShouldBe(3);

        // [0] present row, present cell — REAL infers Float64, so the published CLR type is double.
        snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
        snapshots[0].SourceTimestampUtc.ShouldBe(
            new DateTime(2026, 7, 24, 10, 0, 0, DateTimeKind.Utc));   // from sample_ts, not the poll clock

        // [1] no row at all — NOT the same thing as a NULL cell.
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
        snapshots[1].Value.ShouldBeNull();

        // [2] row present, value cell NULL, nullIsBad = false ⇒ Uncertain (and the row's timestamp survives).
        snapshots[2].Value.ShouldBeNull();
        SqlStatusCodes.IsUncertain(snapshots[2].StatusCode).ShouldBeTrue();
        snapshots[2].SourceTimestampUtc.ShouldBe(
            new DateTime(2026, 7, 24, 10, 0, 1, DateTimeKind.Utc));
    }

    [Fact]
    public async Task ReadAsync_withNullIsBad_publishesBadForANullCell_butStillBadNoDataForAnAbsentRow()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture, nullIsBad: true,
            tags: [KvTag("Temp", SqlitePollFixture.NullValueKey), KvTag("Missing", SqlitePollFixture.AbsentKey)]);

        var snapshots = await reader.ReadAsync(["Temp", "Missing"], CancellationToken.None);

        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Bad);
        SqlStatusCodes.IsBad(snapshots[0].StatusCode).ShouldBeTrue();
        // The nullIsBad switch governs a NULL cell only — an absent row keeps its own, more specific code.
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
    }

    [Fact]
    public async Task ReadAsync_anUnresolvableRef_isBadNodeIdUnknown_andItsNeighboursStillRead()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture,
            KvTag("Speed", SqlitePollFixture.PresentKey),
            KvTag("Pressure", SqlitePollFixture.SecondPresentKey));

        var snapshots = await reader.ReadAsync(
            ["Speed", "NotAuthored", "Pressure"], CancellationToken.None);

        snapshots.Count.ShouldBe(3);
        snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
        // The hole must not shift the tail — this is the ordering contract's real failure mode.
        snapshots[2].Value.ShouldBe(SqlitePollFixture.SecondPresentValue);
    }

    [Fact]
    public async Task ReadAsync_twoTagsSharingOneKeyValue_bothReceiveTheValue()
    {
        // SqlQueryPlan.Members keeps duplicates: these two tags bind ONE parameter but stay TWO members,
        // and a reader that indexed members by key into a 1:1 map would feed only one of them.
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture,
            KvTag("SpeedA", SqlitePollFixture.PresentKey),
            KvTag("SpeedB", SqlitePollFixture.PresentKey));

        var snapshots = await reader.ReadAsync(["SpeedA", "SpeedB"], CancellationToken.None);

        snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
        snapshots[1].Value.ShouldBe(SqlitePollFixture.PresentValue);
        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.Good);
    }

    [Fact]
    public async Task ReadAsync_theSameRefTwice_feedsBothSlots()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey));

        var snapshots = await reader.ReadAsync(["Speed", "Speed"], CancellationToken.None);

        snapshots.Count.ShouldBe(2);
        snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
        snapshots[1].Value.ShouldBe(SqlitePollFixture.PresentValue);
    }

    [Fact]
    public async Task ReadAsync_withNoRefs_returnsAnEmptyList_andOpensNoConnection()
    {
        using var fixture = new SqlitePollFixture();
        var factory = new TrackingFactory();
        var reader = new SqlPollReader(
            factory, fixture.ConnectionString, new SqliteDialect(),
            commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
            maxConcurrentGroups: 4, nullIsBad: false, resolve: _ => null);

        (await reader.ReadAsync([], CancellationToken.None)).ShouldBeEmpty();

        factory.Created.ShouldBe(0);
    }

    // ---- wide row ----

    [Fact]
    public async Task ReadAsync_wideRow_slicesEachColumnToItsOwnMember()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture,
            WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation),
            WideTag("Pressure", "pressure", selectorValue: SqlitePollFixture.PresentStation));

        var snapshots = await reader.ReadAsync(["Oven", "Pressure"], CancellationToken.None);

        // One row, two tags — the wrong-slice defect would cross these two values over.
        snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentStationOvenTemp);
        snapshots[1].Value.ShouldBe(SqlitePollFixture.PresentStationPressure);
        snapshots[0].SourceTimestampUtc.ShouldBe(new DateTime(2026, 7, 24, 10, 0, 0, DateTimeKind.Utc));
    }

    [Fact]
    public async Task ReadAsync_wideRow_absentRowIsBadNoData_andANullCellIsUncertain()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture,
            WideTag("Gone", "oven_temp", selectorValue: SqlitePollFixture.AbsentStation),
            WideTag("NullOven", "oven_temp", selectorValue: SqlitePollFixture.NullOvenTempStation),
            WideTag("NullOvenPressure", "pressure", selectorValue: SqlitePollFixture.NullOvenTempStation));

        var snapshots = await reader.ReadAsync(
            ["Gone", "NullOven", "NullOvenPressure"], CancellationToken.None);

        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);   // no row matched the selector
        SqlStatusCodes.IsUncertain(snapshots[1].StatusCode).ShouldBeTrue();   // row matched, cell NULL
        snapshots[1].Value.ShouldBeNull();
        snapshots[2].Value.ShouldBe(1.6);   // its neighbour on the SAME row is unaffected
    }

    [Fact]
    public async Task ReadAsync_wideRow_topByTimestamp_readsTheNewestRow()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture,
            WideTag("Newest", "oven_temp", topByTimestamp: SqlitePollFixture.TimestampColumn));

        var snapshots = await reader.ReadAsync(["Newest"], CancellationToken.None);

        // Station 8, not the first-inserted station 7 — a lost ORDER BY / row limit shows up here.
        snapshots[0].Value.ShouldBe(SqlitePollFixture.NewestStationOvenTemp);
        snapshots[0].Value.ShouldNotBe(SqlitePollFixture.PresentStationOvenTemp);
    }

    // ---- type + timestamp mapping ----

    [Fact]
    public async Task ReadAsync_coercesTheCellToTheTagsDeclaredType()
    {
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture,
            KvTag("AsInt", SqlitePollFixture.PresentKey, DriverDataType.Int32),
            KvTag("AsText", SqlitePollFixture.PresentKey, DriverDataType.String),
            KvTag("Inferred", SqlitePollFixture.PresentKey));

        var snapshots = await reader.ReadAsync(["AsInt", "AsText", "Inferred"], CancellationToken.None);

        snapshots[0].Value.ShouldBe(42);          // int, not double — the declared type wins
        snapshots[1].Value.ShouldBe("42");        // string
        snapshots[2].Value.ShouldBe(42.0);        // no declaration ⇒ dialect-inferred from REAL
    }

    [Fact]
    public async Task ReadAsync_withNoTimestampColumn_stampsThePollClock()
    {
        using var fixture = new SqlitePollFixture();
        var before = DateTime.UtcNow;
        var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey, timestampColumn: null));

        var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);

        snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
        snapshots[0].SourceTimestampUtc.ShouldNotBeNull();
        snapshots[0].SourceTimestampUtc!.Value.ShouldBeGreaterThanOrEqualTo(before.AddSeconds(-1));
        snapshots[0].SourceTimestampUtc!.Value.ShouldBe(snapshots[0].ServerTimestampUtc);
    }

    [Fact]
    public async Task ReadAsync_whenASourceViolatesOneRowPerKey_takesTheLastRowDeterministically()
    {
        using var fixture = new SqlitePollFixture();
        fixture.Execute(
            $"INSERT INTO {SqlitePollFixture.KeyValueTable} " +
            $"({SqlitePollFixture.KeyColumn}, {SqlitePollFixture.ValueColumn}, {SqlitePollFixture.TimestampColumn}) " +
            $"VALUES ('{SqlitePollFixture.PresentKey}', 99.0, '2026-07-24T10:00:09Z')");
        var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey));

        var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);

        // Design §3.6: the contract is one row per key; when a source breaks it the reader is at least
        // deterministic — last occurrence in reader order — rather than silently arbitrary.
        snapshots[0].Value.ShouldBe(99.0);
    }

    // ---- the frozen-peer contract ----

    [Fact]
    public async Task ReadAsync_whenTheQueryCannotComplete_surfacesBadTimeoutWithinTheDeadline()
    {
        using var fixture = new SqlitePollFixture();

        // A held EXCLUSIVE transaction is this rig's frozen peer: the SELECT below cannot proceed and
        // Microsoft.Data.Sqlite's busy-retry loop is fully SYNCHRONOUS — it neither returns nor honours a
        // cancellation token until CommandTimeout expires. That is precisely the S7 R2-01 shape (an async
        // API that ignores its deadline), so only a real wall-clock bound can end this call early.
        using var locker = fixture.OpenNewConnection();
        using (var begin = locker.CreateCommand())
        {
            begin.CommandText = "BEGIN EXCLUSIVE";
            begin.ExecuteNonQuery();
        }

        try
        {
            // Deliberately inverted against the authoring rule (operationTimeout > commandTimeout): the
            // point is to prove the CLIENT-side bound fires while the server-side backstop would still wait.
            var reader = new SqlPollReader(
                fixture.Factory, fixture.ConnectionString, new SqliteDialect(),
                commandTimeout: TimeSpan.FromSeconds(30), operationTimeout: TimeSpan.FromMilliseconds(500),
                maxConcurrentGroups: 4, nullIsBad: false,
                resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)));

            var clock = Stopwatch.StartNew();
            var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);
            clock.Stop();

            snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout);
            snapshots[0].Value.ShouldBeNull();
            // Without the wall-clock bound this returns at CommandTimeout (30 s), not at 0.5 s.
            clock.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10));
        }
        finally
        {
            using var rollback = locker.CreateCommand();
            rollback.CommandText = "ROLLBACK";
            rollback.ExecuteNonQuery();
        }
    }

    [Fact]
    public async Task ReadAsync_whenTheCallerCancels_propagatesTheCancellation()
    {
        // Deliberate asymmetry: OUR deadline expiring is a data-quality outcome (BadTimeout snapshots, so
        // clients see the staleness); the CALLER cancelling is the engine tearing the poll down, and must
        // propagate rather than be laundered into a snapshot nobody will consume.
        using var fixture = new SqlitePollFixture();
        var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey));
        using var cancelled = new CancellationTokenSource();
        await cancelled.CancelAsync();

        await Should.ThrowAsync<OperationCanceledException>(
            async () => await reader.ReadAsync(["Speed"], cancelled.Token));
    }

    [Fact]
    public async Task ReadAsync_whenTheDatabaseCannotBeOpened_throwsSoTheEngineBacksOff()
    {
        // IReadable's contract: per-tag failures are Bad-coded snapshots, but an unreachable driver throws.
        using var fixture = new SqlitePollFixture();
        var unreachable = new SqliteConnectionStringBuilder
        {
            DataSource = Path.Combine(
                Path.GetTempPath(), $"otopcua-sql-no-such-dir-{Guid.NewGuid():N}", "db.sqlite"),
        }.ToString();
        var reader = new SqlPollReader(
            fixture.Factory, unreachable, new SqliteDialect(),
            commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
            maxConcurrentGroups: 4, nullIsBad: false,
            resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)));

        await Should.ThrowAsync<DbException>(
            async () => await reader.ReadAsync(["Speed"], CancellationToken.None));
    }

    // ---- connection lifecycle ----

    [Theory]
    [InlineData(1)]
    [InlineData(3)]
    public async Task ReadAsync_neverHoldsMoreConnectionsThanMaxConcurrentGroups(int maxConcurrentGroups)
    {
        using var fixture = new SqlitePollFixture();
        // Three distinct group keys — key-value, wide-row where-pair, wide-row topByTimestamp.
        var tags = new[]
        {
            KvTag("Speed", SqlitePollFixture.PresentKey),
            WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation),
            WideTag("Newest", "oven_temp", topByTimestamp: SqlitePollFixture.TimestampColumn),
        };
        // The delay makes the groups genuinely overlap; without it a SQLite query is too fast for
        // concurrency to be observable at all and the cap assertion below would pass vacuously.
        var factory = new TrackingFactory(TimeSpan.FromMilliseconds(150));
        var reader = new SqlPollReader(
            factory, fixture.ConnectionString, new SqliteDialect(),
            commandTimeout: CommandTimeout, operationTimeout: TimeSpan.FromSeconds(30),
            maxConcurrentGroups: maxConcurrentGroups, nullIsBad: false, resolve: Resolver(tags));

        var snapshots = await reader.ReadAsync(["Speed", "Oven", "Newest"], CancellationToken.None);

        snapshots.ShouldAllBe(s => s.StatusCode == SqlStatusCodes.Good);
        factory.Created.ShouldBe(3);
        factory.Peak.ShouldBe(maxConcurrentGroups);   // == 3 for the uncapped run proves the overlap is real
        factory.Live.ShouldBe(0);                     // every connection closed — no leak under the poll loop
    }

    [Fact]
    public async Task ReadAsync_repeatedPolls_leakNoConnections()
    {
        using var fixture = new SqlitePollFixture();
        var factory = new TrackingFactory();
        var reader = new SqlPollReader(
            factory, fixture.ConnectionString, new SqliteDialect(),
            commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
            maxConcurrentGroups: 4, nullIsBad: false,
            resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)));

        for (var poll = 0; poll < 5; poll++)
        {
            var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);
            snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
        }

        factory.Created.ShouldBe(5);
        factory.Live.ShouldBe(0);
    }

    // ---- argument guards ----

    [Fact]
    public void Constructor_rejectsAnUnusableTimeoutOrConcurrencyCap()
    {
        var dialect = new SqliteDialect();
        Should.Throw<ArgumentOutOfRangeException>(() => new SqlPollReader(
            dialect.Factory, "Data Source=x", dialect, TimeSpan.Zero, OperationTimeout, 4, false, _ => null));
        Should.Throw<ArgumentOutOfRangeException>(() => new SqlPollReader(
            dialect.Factory, "Data Source=x", dialect, CommandTimeout, TimeSpan.Zero, 4, false, _ => null));
        Should.Throw<ArgumentOutOfRangeException>(() => new SqlPollReader(
            dialect.Factory, "Data Source=x", dialect, CommandTimeout, OperationTimeout, 0, false, _ => null));
        Should.Throw<ArgumentException>(() => new SqlPollReader(
            dialect.Factory, "  ", dialect, CommandTimeout, OperationTimeout, 4, false, _ => null));
    }

    [Fact]
    public async Task ReadAsync_rejectsANullReferenceList()
        => await Should.ThrowAsync<ArgumentNullException>(async () =>
        {
            using var fixture = new SqlitePollFixture();
            await Reader(fixture).ReadAsync(null!, CancellationToken.None);
        });

    // ---- helpers ----

    private static SqlPollReader Reader(SqlitePollFixture fixture, params SqlTagDefinition[] tags)
        => Reader(fixture, nullIsBad: false, tags: tags);

    private static SqlPollReader Reader(
        SqlitePollFixture fixture, bool nullIsBad, params SqlTagDefinition[] tags)
        => new(fixture.Factory, fixture.ConnectionString, new SqliteDialect(),
            commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
            maxConcurrentGroups: 4, nullIsBad: nullIsBad, resolve: Resolver(tags));

    /// <summary>The driver's RawPath→definition table, standing in for <c>EquipmentTagRefResolver</c>.</summary>
    private static Func<string, SqlTagDefinition?> Resolver(params SqlTagDefinition[] tags)
    {
        var table = tags.ToDictionary(t => t.Name, StringComparer.Ordinal);
        return rawPath => table.GetValueOrDefault(rawPath);
    }

    private static SqlTagDefinition KvTag(
        string name,
        string keyValue,
        DriverDataType? declaredType = null,
        string? timestampColumn = SqlitePollFixture.TimestampColumn)
        => new(name, SqlTagModel.KeyValue, SqlitePollFixture.KeyValueTable,
            KeyColumn: SqlitePollFixture.KeyColumn, KeyValue: keyValue,
            ValueColumn: SqlitePollFixture.ValueColumn,
            TimestampColumn: timestampColumn,
            DeclaredType: declaredType);

    private static SqlTagDefinition WideTag(
        string name,
        string columnName,
        string? selectorValue = null,
        string? topByTimestamp = null,
        string? timestampColumn = SqlitePollFixture.TimestampColumn)
        => new(name, SqlTagModel.WideRow, SqlitePollFixture.WideRowTable,
            ColumnName: columnName,
            TimestampColumn: timestampColumn,
            RowSelectorColumn: selectorValue is null ? null : SqlitePollFixture.WideRowSelectorColumn,
            RowSelectorValue: selectorValue,
            RowSelectorTopByTimestamp: topByTimestamp);

    /// <summary>
    ///     A <see cref="DbProviderFactory"/> over SQLite that counts how many connections the reader has
    ///     created and how many are alive at once — the only way to observe the connection lifecycle from
    ///     outside, since <see cref="SqlPollReader"/> deliberately owns its connections end to end.
    ///     <para>The optional <c>createDelay</c> widens the window each connection is alive so that
    ///     concurrent group execution is actually observable against a database whose queries would
    ///     otherwise finish in microseconds.</para>
    /// </summary>
    private sealed class TrackingFactory(TimeSpan createDelay = default) : DbProviderFactory
    {
        private readonly object _sync = new();
        private int _live;
        private int _peak;
        private int _created;

        /// <summary>How many connections the reader has asked the factory for.</summary>
        public int Created { get { lock (_sync) { return _created; } } }

        /// <summary>How many created connections have not yet closed.</summary>
        public int Live { get { lock (_sync) { return _live; } } }

        /// <summary>The high-water mark of <see cref="Live"/>.</summary>
        public int Peak { get { lock (_sync) { return _peak; } } }

        public override DbConnection CreateConnection()
        {
            var connection = SqliteFactory.Instance.CreateConnection()!;
            connection.StateChange += OnStateChange;
            lock (_sync)
            {
                _created++;
                _live++;
                if (_live > _peak) _peak = _live;
            }

            if (createDelay > TimeSpan.Zero) Thread.Sleep(createDelay);
            return connection;
        }

        private void OnStateChange(object sender, StateChangeEventArgs e)
        {
            if (e.CurrentState != ConnectionState.Closed && e.CurrentState != ConnectionState.Broken) return;
            lock (_sync) { _live--; }
        }
    }
}