lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs

using System.Data;
using System.Data.Common;
using System.Diagnostics;
using System.Globalization;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
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);
    }

    [Fact]
    public async Task ReadAsync_wideRow_whenTheSelectorMatchesManyRows_warnsInsteadOfSilentlyPickingOne()
    {
        using var fixture = new SqlitePollFixture();
        // A second row for the SAME station. The where-pair form emits no ORDER BY and no row limit, so
        // which of the two the reader publishes is decided by physical storage order — an authoring mistake
        // (a selector column that is not actually unique) that must not pass silently. This is the WideRow
        // counterpart of ReadAsync_whenASourceViolatesOneRowPerKey_...'s duplicate-key warning.
        AddSecondRowForPresentStation(fixture);
        var logger = new RecordingLogger();
        var reader = Reader(fixture, logger,
            WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation));

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

        // Still a value — the reader degrades to "last row wins", exactly as the key-value model does.
        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);

        var warning = logger.Entries.ShouldHaveSingleItem();
        warning.Level.ShouldBe(LogLevel.Warning);
        // Names the table AND the selector, so an operator can find the offending tag from the log alone.
        warning.Message.ShouldContain(SqlitePollFixture.WideRowTable);
        warning.Message.ShouldContain(SqlitePollFixture.WideRowSelectorColumn);
        warning.Message.ShouldContain(SqlitePollFixture.PresentStation);
    }

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

        // A selector that matches exactly one row is the normal case and must log nothing at all —
        // without this leg a warning that always fires would still pass the test above.
        var clean = Reader(fixture, logger,
            WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation));
        await clean.ReadAsync(["Oven"], CancellationToken.None);
        logger.Entries.ShouldBeEmpty();

        AddSecondRowForPresentStation(fixture);
        var ambiguous = Reader(fixture, logger,
            WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation));
        await ambiguous.ReadAsync(["Oven"], CancellationToken.None);
        await ambiguous.ReadAsync(["Oven"], CancellationToken.None);

        // A poll loop hits this every cycle; the rate limit is what stops a misconfigured table flooding
        // the log. Two polls, one warning.
        logger.Entries.Count.ShouldBe(1);
    }

    // ---- 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();
        Execute(locker, "BEGIN EXCLUSIVE");

        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
        {
            Execute(locker, "ROLLBACK");
        }
    }

    [Fact]
    public async Task ReadAsync_aTimedOutGroupKeepsItsConcurrencySlotUntilTheQueryTrulyFinishes()
    {
        // The reason the concurrency slot is released by the WORK and not by the waiter: a timed-out group
        // is still holding a connection, so handing its slot back at the deadline would let the next poll
        // open another one — and a database that stays frozen would accumulate one connection per poll pass
        // forever. This test is the only thing that distinguishes the two designs; the concurrency-cap test
        // above never times a group out, and the BadTimeout test above never counts connections.
        using var fixture = new SqlitePollFixture();
        var factory = new TrackingFactory();
        var reader = new SqlPollReader(
            factory, fixture.ConnectionString, new SqliteDialect(),
            // commandTimeout deliberately far beyond the test's own horizon: the wedged query must stay
            // wedged for as long as the lock is held, so nothing but the test releases the slot.
            commandTimeout: TimeSpan.FromSeconds(30), operationTimeout: TimeSpan.FromMilliseconds(500),
            maxConcurrentGroups: 1, nullIsBad: false,
            resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)));

        var locker = fixture.OpenNewConnection();
        Execute(locker, "BEGIN EXCLUSIVE");   // outside the try: a ROLLBACK with no open transaction throws
        try
        {
            // Poll 1 wedges against the lock and times out, abandoning a still-running query.
            var first = await reader.ReadAsync(["Speed"], CancellationToken.None);
            first[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout);
            factory.Created.ShouldBe(1);
            factory.Live.ShouldBe(1);   // the zombie still owns its connection

            // (a) The slot is STILL held. maxConcurrentGroups is 1, so poll 2 cannot even reach the
            // factory: it times out waiting on the gate. Created staying at 1 is the load-bearing
            // assertion — release-from-the-waiter would make this 2.
            var second = await reader.ReadAsync(["Speed"], CancellationToken.None);
            second[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout);
            factory.Created.ShouldBe(1);
            factory.Live.ShouldBe(1);
        }
        finally
        {
            Execute(locker, "ROLLBACK");
            locker.Dispose();
        }

        // (b) With the lock gone the zombie's query completes (or faults on its own cancelled deadline),
        // closes its connection, and only then hands the slot back.
        (await WaitUntilAsync(() => factory.Live == 0, TimeSpan.FromSeconds(30)))
            .ShouldBeTrue("the abandoned group never released its connection");

        var third = await reader.ReadAsync(["Speed"], CancellationToken.None);
        third[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
        factory.Created.ShouldBe(2);   // the slot was reusable exactly once the zombie finished
        factory.Live.ShouldBe(0);
    }

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

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

    /// <summary>
    ///     Adds a second <see cref="SqlitePollFixture.WideRowTable"/> row for the station the wide-row tests
    ///     select, breaking that model's one-row-per-selector contract.
    /// </summary>
    private static void AddSecondRowForPresentStation(SqlitePollFixture fixture)
        => fixture.Execute(string.Create(CultureInfo.InvariantCulture, $"""
            INSERT INTO {SqlitePollFixture.WideRowTable}
                ({SqlitePollFixture.WideRowSelectorColumn}, oven_temp, pressure,
                 {SqlitePollFixture.TimestampColumn})
            VALUES ({SqlitePollFixture.PresentStation}, 999.0, 9.9, '2026-07-24T10:00:09Z')
            """));

    /// <summary>Runs one statement on a caller-owned connection.</summary>
    private static void Execute(SqliteConnection connection, string sql)
    {
        using var command = connection.CreateCommand();
        command.CommandText = sql;
        command.ExecuteNonQuery();
    }

    /// <summary>
    ///     Polls <paramref name="condition"/> until it holds or <paramref name="limit"/> elapses. Bounded on
    ///     purpose: a test that asserts an abandoned task eventually finishes must fail, not hang, when it
    ///     does not.
    /// </summary>
    private static async Task<bool> WaitUntilAsync(Func<bool> condition, TimeSpan limit)
    {
        var clock = Stopwatch.StartNew();
        while (clock.Elapsed < limit)
        {
            if (condition()) return true;
            await Task.Delay(25);
        }

        return condition();
    }

    /// <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>
    ///     Captures the reader's log so "reported, not silent" can be asserted rather than assumed — the
    ///     contract-violation warnings are the reader's only signal that a source is misauthored, so they are
    ///     behaviour, not diagnostics.
    /// </summary>
    private sealed class RecordingLogger : ILogger
    {
        /// <summary>Every record written, level + rendered message.</summary>
        public List<(LogLevel Level, string Message)> Entries { get; } = [];

        public IDisposable BeginScope<TState>(TState state) where TState : notnull => Scope.Instance;

        public bool IsEnabled(LogLevel logLevel) => true;

        public void Log<TState>(
            LogLevel logLevel, EventId eventId, TState state, Exception? exception,
            Func<TState, Exception?, string> formatter)
        {
            lock (Entries) Entries.Add((logLevel, formatter(state, exception)));
        }

        private sealed class Scope : IDisposable
        {
            public static Scope Instance { get; } = new();

            public void Dispose() { }
        }
    }

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