lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs

using System.Data.Common;
using System.Diagnostics.CodeAnalysis;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;

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

/// <summary>
///     The <c>Sql</c> driver instance: a <b>read-only</b> Equipment-kind driver that polls SQL tables/views
///     and publishes selected cells as OPC UA variables. Structurally the SQL analogue of
///     <c>ModbusDriver</c> — this type owns lifecycle, the authored RawPath table, health, host
///     connectivity, and the <see cref="PollGroupEngine"/> subscription overlay, while
///     <see cref="SqlPollReader"/> owns the read path.
///     <para><b>Read-only is structural, not configured.</b> <see cref="IWritable"/> is deliberately not
///     implemented, so no amount of config (and no <c>WriteOperate</c> role) can produce a write; every
///     discovered variable is <see cref="SecurityClassification.ViewOnly"/>. A future write feature is a
///     new capability, not a flag flip.</para>
///     <para><b>Health is classified here, because nothing below does it.</b> The reader honours
///     <see cref="IReadable"/> literally: it throws only when the database cannot be reached, and turns
///     every other failure — including a frozen database — into Bad-coded snapshots. Those return
///     <em>successfully</em>, so <see cref="PollGroupEngine"/> sees a clean tick, does not back off, and does
///     not call <see cref="HandlePollError"/>. Without <see cref="ObservePollOutcome"/> below, a driver whose
///     every value is <see cref="SqlStatusCodes.BadTimeout"/> would keep reporting
///     <see cref="DriverState.Healthy"/>. See that method for the exact rule and for why it does not
///     manufacture an exception to force backoff.</para>
/// </summary>
public sealed class SqlDriver
    : IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe, ISkippedTagReporter, IAsyncDisposable
{
    /// <summary>Authored tags this driver could not build into its poll table (Gitea #541).</summary>
    private readonly SkippedTagLedger _skippedTags = new();

    /// <summary>
    ///     Blob-level skips from the most recent <see cref="BuildTagTable"/>, retained because Sql drops a
    ///     tag at TWO points and the report must be the union of both.
    /// </summary>
    /// <remarks>
    ///     A tag can fail to map from its blob (<see cref="BuildTagTable"/>) OR be rejected against the live
    ///     catalog (<see cref="ApplyCatalogGateAsync"/>) — a renamed column, a dropped view. Both leave the
    ///     tag out of the polled table publishing <c>BadNodeIdUnknown</c> with the driver Healthy, so both
    ///     are skips. Reporting only the first would make a Sql driver that lost half its tags to schema
    ///     drift claim "0 skipped", which is worse than saying nothing at all.
    /// </remarks>
    private IReadOnlyList<SkippedTag> _blobSkips = [];

    /// <inheritdoc />
    public SkippedTagReport? GetSkippedTags() => _skippedTags.Snapshot();

    /// <summary>
    ///     Republishes the skip ledger as the union of the blob-level skips and the catalog gate's
    ///     rejections, against the true authored denominator (<c>RawTags.Count</c>, not the post-mapping
    ///     table size — a tag that failed to map is still a tag the operator authored).
    /// </summary>
    /// <param name="gateRejections">Catalog-gate rejections; empty before the gate has run.</param>
    private void PublishSkipLedger(IEnumerable<SkippedTag> gateRejections)
    {
        _skippedTags.BeginRebuild(_options.RawTags.Count);
        foreach (var skip in _blobSkips) _skippedTags.Skip(skip.RawPath, skip.Reason);
        foreach (var skip in gateRejections) _skippedTags.Skip(skip.RawPath, skip.Reason);
        _skippedTags.Commit();
    }

    /// <summary>
    ///     The driver-type string this driver registers under — <see cref="DriverTypeNames.Sql"/>, the single
    ///     source of truth the deploy pipeline stores in <c>DriverInstance.DriverType</c> and every dispatch
    ///     map keys by. Chained off the shared constant (Task 11 wired the factory + probe, so
    ///     <c>DriverTypeNamesGuardTests</c>' bidirectional-parity check is satisfied).
    /// </summary>
    public const string DriverTypeName = DriverTypeNames.Sql;

    /// <summary>Shown for a connection string that names no recognisable server.</summary>
    private const string UnknownEndpoint = "(unknown sql endpoint)";

    /// <summary>Upper bound on the poll-loop failure backoff — the S7-proven 30 s cap adopted fleet-wide (05/STAB-8).</summary>
    private static readonly TimeSpan PollBackoffCap = TimeSpan.FromSeconds(30);

    /// <summary>Connection-string keys that name the server, in the order they are consulted.</summary>
    private static readonly string[] ServerKeys =
        ["server", "data source", "datasource", "host", "address", "addr", "network address"];

    /// <summary>Connection-string keys that name the database.</summary>
    private static readonly string[] DatabaseKeys = ["database", "initial catalog"];

    // ---- instance fields (grouped at top for auditability) ----

    /// <summary>Config-derived, so mutable — see <see cref="ApplyBinding"/>.</summary>
    private SqlDriverOptions _options;
    private readonly string _driverInstanceId;
    /// <summary>Config-derived, so mutable: <see cref="ApplyBinding"/> swaps it with the connection string
    /// and options together on a reinitialize (#516).</summary>
    private ISqlDialect _dialect;
    private DbProviderFactory _factory;

    /// <summary>
    ///     The resolved connection string — a <b>secret</b>. It is passed to the provider and to nothing
    ///     else: every log line, health message and host status carries <see cref="Endpoint"/> instead.
    /// </summary>
    private string _connectionString;

    private readonly ILogger<SqlDriver> _logger;

    /// <summary>
    ///     The authored RawPath → definition table, held as an <b>immutable snapshot swapped atomically</b>
    ///     rather than as a mutated dictionary (the shape <c>ModbusDriver</c> uses).
    ///     <see cref="ReinitializeAsync"/> rebuilds this table while poll loops are reading it; clearing and
    ///     refilling a shared <see cref="Dictionary{TKey,TValue}"/> under that concurrency is a torn read at
    ///     best and an <see cref="InvalidOperationException"/> inside a poll at worst.
    /// </summary>
    private IReadOnlyDictionary<string, SqlTagDefinition> _tagsByRawPath =
        new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal);

    /// <summary>
    ///     The subset of <see cref="_tagsByRawPath"/> that survived the design §8.1 catalog gate, with every
    ///     identifier rewritten to the catalog's own spelling. <b>This — not the authored table — is what a
    ///     read or a poll resolves against</b>, so an identifier that never appeared in the catalog can never
    ///     reach a query.
    ///     <para><b>Why the two tables are separate.</b> A tag the gate rejected must still <em>exist</em> as a
    ///     node: §8.1's specified outcome is that it publishes
    ///     <see cref="SqlStatusCodes.BadNodeIdUnknown"/>, and a status code can only be published by a node
    ///     that is there. Dropping rejected tags from the authored table too would delete the node instead,
    ///     turning a diagnosable Bad quality into a silently missing address-space entry — a strictly worse
    ///     failure for the operator trying to find their typo.</para>
    ///     <para>Swapped atomically, for the same reason the authored table is.</para>
    /// </summary>
    private IReadOnlyDictionary<string, SqlTagDefinition> _polledByRawPath =
        new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal);

    /// <summary>Resolves a read/subscribe RawPath to its <b>catalog-validated</b> definition, or a miss.</summary>
    private readonly EquipmentTagRefResolver<SqlTagDefinition> _resolver;

    /// <summary>The factory the CALLER passed, or null to take the dialect's. Kept separate from
    /// <see cref="_factory"/> so a re-derived dialect cannot silently displace a test's injected factory.</summary>
    private readonly DbProviderFactory? _explicitFactory;

    /// <summary>Re-parses a <c>DriverConfig</c> into every config-derived dependency. Null when the driver
    /// was constructed directly, which keeps the constructor-supplied trio authoritative.</summary>
    private readonly Func<string, SqlDriverBinding>? _rebind;

    /// <summary>
    ///     The read path. Built once, in the constructor, and <b>not</b> injected: its <c>resolve</c>
    ///     delegate must read this driver's live authored table, which does not exist before the driver does.
    /// </summary>
    private SqlPollReader _reader;

    /// <summary>Polled subscriptions. The driver supplies the reader + change bridge; the engine owns the loop.</summary>
    private readonly PollGroupEngine _poll;

    // Single logical host: one driver instance dials one connection string. HostName is the endpoint
    // description (server[/database]) so the Admin UI shows operators where to look.
    private readonly object _probeLock = new();
    private HostState _hostState = HostState.Unknown;
    private DateTime _hostStateChangedUtc = DateTime.UtcNow;

    private DriverHealth _health = new(DriverState.Unknown, null, null);

    /// <summary>
    ///     The exception instance <see cref="ReadAsync"/> has already turned into a health verdict, held only
    ///     until <see cref="HandlePollError"/> — which the poll engine calls with that same object — has
    ///     consumed it (Driver.Sql-004).
    /// </summary>
    /// <remarks>
    ///     A field rather than a flag on the exception itself: <see cref="Exception.Data"/> is provider-owned
    ///     and may be read-only, and wrapping the throw would change the type the engine and
    ///     <see cref="IReadable"/>'s callers see. Concurrent poll loops can overwrite each other's mark; the
    ///     loser is then merely re-degraded with a poorer message, which is the pre-fix behaviour rather than
    ///     a new failure.
    /// </remarks>
    private Exception? _classifiedByRead;

    /// <summary>Occurs when a subscribed tag's value or quality changes.</summary>
    public event EventHandler<DataChangeEventArgs>? OnDataChange;

    /// <summary>Occurs when the database endpoint transitions between reachable and unreachable.</summary>
    public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;

    // ---- ctor + identity ----

    /// <summary>Initializes a new <c>Sql</c> driver instance.</summary>
    /// <param name="options">The driver's typed configuration (the factory applies the documented defaults).</param>
    /// <param name="driverInstanceId">The central config DB's identity for this driver instance.</param>
    /// <param name="dialect">
    ///     The provider seam — identifier quoting, liveness SQL, row-limit syntax and column-type mapping.
    /// </param>
    /// <param name="connectionString">
    ///     The <b>already-resolved</b> connection string. The driver never resolves a
    ///     <c>connectionStringRef</c> itself and never logs this value.
    /// </param>
    /// <param name="factory">
    ///     Creates the provider's connections. Defaults to <paramref name="dialect"/>'s factory; passed
    ///     explicitly by tests that need to observe or delay connection creation.
    /// </param>
    /// <param name="logger">Optional; defaults to a no-op logger.</param>
    /// <param name="rebind">
    ///     Optional re-parse of a <c>DriverConfig</c> into every config-derived dependency, supplied by the
    ///     factory so <see cref="InitializeAsync"/> can adopt a changed config. Null (direct construction —
    ///     every test) keeps the constructor-supplied options, dialect and connection string.
    /// </param>
    /// <exception cref="ArgumentNullException">A required reference argument is null.</exception>
    /// <exception cref="ArgumentException"><paramref name="driverInstanceId"/> or <paramref name="connectionString"/> is blank.</exception>
    public SqlDriver(
        SqlDriverOptions options,
        string driverInstanceId,
        ISqlDialect dialect,
        string connectionString,
        DbProviderFactory? factory = null,
        ILogger<SqlDriver>? logger = null,
        Func<string, SqlDriverBinding>? rebind = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(dialect);
        ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
        if (string.IsNullOrWhiteSpace(connectionString))
            throw new ArgumentException("A Sql driver needs a resolved connection string.", nameof(connectionString));

        _driverInstanceId = driverInstanceId;
        _logger = logger ?? NullLogger<SqlDriver>.Instance;
        _explicitFactory = factory;
        _rebind = rebind;
        _resolver = new EquipmentTagRefResolver<SqlTagDefinition>(Lookup);

        // Everything below this line is derived from config and is re-derived wholesale on a reinitialize.
        ApplyBinding(options, dialect, connectionString);

        _poll = new PollGroupEngine(
            reader: ReadAsync,
            onChange: (handle, tagRef, snapshot) =>
                OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)),
            onError: HandlePollError,
            backoffCap: PollBackoffCap);
    }

    /// <summary>
    ///     Adopts a config-derived trio — options, dialect, connection string — and rebuilds everything
    ///     downstream of it (the provider factory, the endpoint description, and the poll reader, which
    ///     captures all three). Called from the constructor and again from <see cref="InitializeAsync"/>
    ///     when a changed config arrives.
    ///     <para><b>All-or-nothing on purpose.</b> The reason Sql was excluded from the first #516 pass is
    ///     that adopting options alone would poll a NEW tag set through the OLD connection and dialect. Doing
    ///     it in one method means a future field derived from config gets added here rather than being
    ///     forgotten on the reinit path.</para>
    ///     <para>Not thread-safe by design: only the init path calls it, before any poll group is running.
    ///     The poll engine and the tag resolver survive across a rebind — the engine holds a delegate to
    ///     <c>ReadAsync</c> and the resolver reads the live tag table, so neither captures the old trio.</para>
    /// </summary>
    /// <param name="options">The typed options to adopt.</param>
    /// <param name="dialect">The provider seam to adopt.</param>
    /// <param name="connectionString">The resolved connection string to adopt. Never logged.</param>
    [MemberNotNull(nameof(_options), nameof(_dialect), nameof(_factory), nameof(_connectionString), nameof(_reader))]
    private void ApplyBinding(SqlDriverOptions options, ISqlDialect dialect, string connectionString)
    {
        _options = options;
        _dialect = dialect;
        _factory = _explicitFactory ?? dialect.Factory;
        _connectionString = connectionString;
        Endpoint = DescribeEndpoint(connectionString);
        _reader = new SqlPollReader(
            _factory,
            connectionString,
            dialect,
            commandTimeout: options.CommandTimeout,
            operationTimeout: options.OperationTimeout,
            maxConcurrentGroups: options.MaxConcurrentGroups,
            nullIsBad: options.NullIsBad,
            resolve: rawPath => _resolver.TryResolve(rawPath, out var definition) ? definition : null,
            logger: _logger);
    }

    /// <summary>True when the supplied DriverConfig JSON carries a real body. The bootstrapper always passes
    /// a populated document; some unit tests pass <c>"{}"</c> or an empty string to exercise lifecycle shape
    /// without a config — those keep the constructor-supplied trio.</summary>
    private static bool HasConfigBody(string? driverConfigJson)
    {
        if (string.IsNullOrWhiteSpace(driverConfigJson)) return false;
        var trimmed = driverConfigJson.Trim();
        return trimmed is not "{}" and not "[]";
    }

    /// <inheritdoc />
    public string DriverInstanceId => _driverInstanceId;

    /// <inheritdoc />
    public string DriverType => DriverTypeName;

    /// <summary>
    ///     The credential-free description of the database this instance polls — <c>server/database</c> where
    ///     the connection string names both. This is the only rendering of the connection string that may
    ///     appear in a log, a health message, or the Admin UI.
    /// </summary>
    public string Endpoint { get; private set; } = "";

    /// <summary>Host identifier surfaced through <see cref="IHostConnectivityProbe"/> — the endpoint description.</summary>
    public string HostName => Endpoint;

    /// <summary>Active polled-subscription count. Diagnostics + disposal assertions.</summary>
    internal int ActiveSubscriptionCount => _poll.ActiveSubscriptionCount;

    /// <summary>The current authored-tag snapshot. Read through a barrier — <see cref="BuildTagTable"/> swaps it.</summary>
    private IReadOnlyDictionary<string, SqlTagDefinition> Tags => Volatile.Read(ref _tagsByRawPath);

    /// <summary>
    ///     The catalog-validated snapshot. Read through a barrier — <see cref="ApplyCatalogGateAsync"/> swaps it.
    /// </summary>
    private IReadOnlyDictionary<string, SqlTagDefinition> PolledTags => Volatile.Read(ref _polledByRawPath);

    /// <summary>
    ///     The resolver's lookup: RawPath → <b>catalog-validated</b> definition, or null on a miss.
    ///     <para>Deliberately reads <see cref="PolledTags"/> rather than <see cref="Tags"/>: a tag the gate
    ///     rejected must miss here so the reader publishes
    ///     <see cref="SqlStatusCodes.BadNodeIdUnknown"/>, exactly as design §8.1 specifies.</para>
    /// </summary>
    private SqlTagDefinition? Lookup(string rawPath) => PolledTags.GetValueOrDefault(rawPath);

    // ---- IDriver lifecycle ----

    /// <summary>
    ///     Builds the authored RawPath table and proves the database is reachable.
    ///     <para><paramref name="driverConfigJson"/> <b>is</b> re-parsed here when the driver was built by
    ///     the factory (Gitea #516). The original comment on this method claimed it was not, on the premise
    ///     that "config parsing belongs to the factory, which builds a fresh instance" — that was false when
    ///     written, because the host reinitialized the EXISTING child in place and every config edit was
    ///     silently discarded while the deployment still sealed green.</para>
    ///     <para>The re-parse is <b>all-or-nothing</b> via <see cref="ApplyBinding"/>: options, the
    ///     <see cref="ISqlDialect"/> and the resolved connection string are re-derived by one
    ///     <c>ParseBinding</c> call and adopted together. That is what makes it safe here — adopting options
    ///     alone would poll a NEW tag set through the OLD database, which is why this driver was excluded
    ///     from the first #516 pass. It happens BEFORE <c>BuildTagTable</c>, so the tag table and the
    ///     connection it is polled over always come from the same revision.</para>
    ///     <para>A driver constructed directly (every unit test) gets no rebinder and keeps its
    ///     constructor-supplied trio, so <c>"{}"</c>-passing lifecycle tests are unaffected.
    ///     <c>DriverSpawnPlanner</c>'s stop + respawn remains the outer guarantee.</para>
    ///     <para>The table is built first because it is pure and cannot fail; the liveness check is the only
    ///     I/O, and on failure this method records <see cref="DriverState.Faulted"/> <b>and rethrows</b> —
    ///     <c>DriverInstanceActor</c> reads a throw as InitializeFailed and lands in Reconnecting with its
    ///     retry timer running, which is exactly the recovery a database that is merely down needs.</para>
    ///     <para>The liveness check is then followed by the design §8.1 <b>catalog gate</b>
    ///     (<see cref="ApplyCatalogGateAsync"/>), which is the second and last piece of I/O. It runs after
    ///     liveness because it needs a working connection, and before the driver reports Healthy because the
    ///     tag table it publishes must be the validated one — a poll must never see an unvalidated
    ///     identifier.</para>
    /// </summary>
    /// <param name="driverConfigJson">The driver configuration JSON (unused; see remarks).</param>
    /// <param name="cancellationToken">Cancellation token for the operation.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    /// <exception cref="InvalidOperationException">The database could not be reached, or its catalog could not be read.</exception>
    public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        WriteHealth(new DriverHealth(DriverState.Initializing, null, null));

        // #516: adopt a changed config BEFORE the tag table is built, so the table and the connection it
        // will be polled over come from the same revision. ApplyBinding takes the options, dialect and
        // connection string as one unit; adopting options alone would poll a NEW tag set through the OLD
        // database, which is why this driver was excluded from the first pass. A null rebinder (direct
        // construction) or an empty/placeholder document keeps the constructor-supplied trio.
        if (_rebind is not null && HasConfigBody(driverConfigJson))
        {
            var binding = _rebind(driverConfigJson);
            ApplyBinding(binding.Options, binding.Dialect, binding.ConnectionString);
        }

        BuildTagTable();

        try
        {
            await VerifyLivenessAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // The caller tore the initialization down; not a database verdict.
            WriteHealth(new DriverHealth(DriverState.Unknown, null, null));
            throw;
        }
        catch (Exception ex)
        {
            // Never interpolate _connectionString OR ex.Message onto the operator surface. The driver is
            // dialect-agnostic over an arbitrary DbProviderFactory, so it cannot rely on any one provider's
            // message discipline — Microsoft.Data.SqlClient's connection-string parser, for one, echoes an
            // unrecognised keyword lower-cased, which a value-based password redactor misses entirely. Only
            // the exception TYPE reaches LastError/host status; the full exception object goes to the log
            // SINK via the structured logger's exception parameter (below), never into a rendered string.
            var message =
                $"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.GetType().Name}. " +
                $"Check the database is reachable and the connection string is valid.";
            _logger.LogError(ex, "Sql driver {DriverInstanceId} liveness check against {Endpoint} failed.",
                _driverInstanceId, Endpoint);
            WriteHealth(new DriverHealth(DriverState.Faulted, null, message));
            TransitionHostTo(HostState.Stopped);
            throw new InvalidOperationException(message, ex);
        }

        // Separate try from liveness so the operator surface names the stage that actually failed: "could
        // not reach the database" and "reached it but could not read its catalog" send an operator to
        // different places, and the second is usually a GRANT rather than an outage.
        try
        {
            await ApplyCatalogGateAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            WriteHealth(new DriverHealth(DriverState.Unknown, null, null));
            throw;
        }
        catch (Exception ex)
        {
            // Same discipline as above: exception TYPE only on the operator surface, full exception to the
            // log sink. A catalog read that fails is NOT evidence the tags are wrong — it is the absence of
            // evidence — so this faults the driver (and retries) rather than rejecting every tag and serving
            // a confidently-empty address space.
            var message =
                $"Sql driver '{_driverInstanceId}' reached {Endpoint} but could not read its catalog to " +
                $"validate authored tables/columns: {ex.GetType().Name}. Check that the connecting principal " +
                "can read the catalog views.";
            _logger.LogError(ex,
                "Sql driver {DriverInstanceId} catalog validation against {Endpoint} failed.",
                _driverInstanceId, Endpoint);
            WriteHealth(new DriverHealth(DriverState.Faulted, null, message));
            TransitionHostTo(HostState.Stopped);
            throw new InvalidOperationException(message, ex);
        }

        WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
        TransitionHostTo(HostState.Running);
        _logger.LogInformation(
            "Sql driver {DriverInstanceId} initialized against {Endpoint} with {TagCount} authored tag(s).",
            _driverInstanceId, Endpoint, Tags.Count);
    }

    /// <inheritdoc />
    public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        await ShutdownAsync(cancellationToken).ConfigureAwait(false);
        await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
    }

    /// <inheritdoc />
    public async Task ShutdownAsync(CancellationToken cancellationToken)
    {
        var lastRead = ReadHealth().LastSuccessfulRead;
        await TeardownAsync().ConfigureAwait(false);
        WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null));
    }

    /// <inheritdoc />
    public DriverHealth GetHealth() => ReadHealth();


    // ---- ITagDiscovery ----

    /// <summary>
    ///     Exactly one pass: the tag set is the authored one, fixed at Initialize, so re-running discovery
    ///     could only produce the same nodes.
    /// </summary>
    public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;

    /// <summary>
    ///     False — <see cref="DiscoverAsync"/> replays authored tags rather than enumerating the backend, so
    ///     the universal discovery browser must not treat this driver as browsable. (Live schema browsing is
    ///     a separate surface: <c>Driver.Sql.Browser</c>.)
    /// </summary>
    public bool SupportsOnlineDiscovery => false;

    /// <summary>
    ///     Streams the authored tags as read-only variables.
    ///     <para>An undeclared tag materialises as <see cref="DriverDataType.String"/> — the same fallback
    ///     <see cref="ISqlDialect.MapColumnType"/> uses for an unrecognised column type — because a column's
    ///     real type is only known once a poll has returned result-set metadata, which has not happened at
    ///     discovery time. Author <c>"type"</c> on a numeric tag.</para>
    /// </summary>
    /// <param name="builder">The address-space builder to stream discovered nodes into.</param>
    /// <param name="cancellationToken">Cancellation token for the operation.</param>
    /// <returns>A completed task — discovery is synchronous over the authored table.</returns>
    public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(builder);
        var folder = builder.Folder(DriverTypeName, DriverTypeName);
        foreach (var tag in Tags.Values)
        {
            if (tag.DeclaredType is null)
            {
                // Discovery cannot infer a column's real type — that needs live result-set metadata, which
                // only a poll returns — so an omitted-type tag materialises as String (the same fallback
                // ISqlDialect.MapColumnType uses) while the reader coerces and publishes the column's actual,
                // possibly numeric, CLR value against that String node. Nothing else signals the operator to
                // this declared-vs-published mismatch, so warn and point them at the fix.
                _logger.LogWarning(
                    "Sql tag '{Tag}' has no authored \"type\" and is discovered as String; a numeric column " +
                    "will then publish numeric values against a String node. Author an explicit \"type\" on " +
                    "numeric tags. Driver={DriverInstanceId}",
                    tag.Name, _driverInstanceId);
            }

            folder.Variable(tag.Name, tag.Name, new DriverAttributeInfo(
                FullName: tag.Name,
                DriverDataType: tag.DeclaredType ?? DriverDataType.String,
                IsArray: false,
                ArrayDim: null,
                // v1 is read-only: no tag, and no configuration, can widen this.
                SecurityClass: SecurityClassification.ViewOnly,
                IsHistorized: false,
                IsAlarm: false,
                WriteIdempotent: false));
        }

        return Task.CompletedTask;
    }

    // ---- IReadable ----

    /// <summary>
    ///     Reads a batch, delegating to <see cref="SqlPollReader"/> and classifying what came back
    ///     (<see cref="ObservePollOutcome"/>).
    /// </summary>
    /// <param name="fullReferences">The RawPaths to read.</param>
    /// <param name="cancellationToken">Cancellation token for the operation.</param>
    /// <returns>One snapshot per reference, at the same index.</returns>
    /// <exception cref="DbException">The database could not be reached — the engine backs off on this.</exception>
    public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
        IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
    {
        // Guarded here rather than inside the try, so a caller's bug cannot be misread as "the database is
        // unreachable" and degrade a perfectly healthy driver.
        ArgumentNullException.ThrowIfNull(fullReferences);

        try
        {
            var snapshots = await _reader.ReadAsync(fullReferences, cancellationToken).ConfigureAwait(false);
            ObservePollOutcome(snapshots);
            return snapshots;
        }
        catch (OperationCanceledException)
        {
            // Teardown, not a database verdict — leave health and host state exactly as they were.
            throw;
        }
        catch (Exception ex)
        {
            // The reader throws for one reason only: opening a connection failed (IReadable's contract).
            _logger.LogWarning(ex, "Sql driver {DriverInstanceId} could not reach {Endpoint} during a read.",
                _driverInstanceId, Endpoint);
            // Type name only — never ex.Message (see InitializeAsync for why); the full exception is already
            // on the log sink via the LogWarning exception parameter above.
            Degrade($"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.GetType().Name}.");
            TransitionHostTo(HostState.Stopped);
            // Mark THIS exception object as already classified before it goes back to the poll engine, which
            // routes it to HandlePollError next. The mark is what makes the deferral there exact — see that
            // method for why inferring it from the exception's type was not (Driver.Sql-004).
            Volatile.Write(ref _classifiedByRead, ex);
            throw;
        }
    }

    // ---- ISubscribable (polling overlay via the shared engine) ----

    /// <summary>
    ///     Registers a polled subscription.
    ///     <para>A non-positive <paramref name="publishingInterval"/> falls back to the configured
    ///     <see cref="SqlDriverOptions.DefaultPollInterval"/>; anything faster than
    ///     <see cref="PollGroupEngine.DefaultMinInterval"/> is floored by the engine.</para>
    /// </summary>
    /// <param name="fullReferences">The RawPaths to poll.</param>
    /// <param name="publishingInterval">The requested publishing interval.</param>
    /// <param name="cancellationToken">Cancellation token for the operation.</param>
    /// <returns>The subscription handle.</returns>
    public Task<ISubscriptionHandle> SubscribeAsync(
        IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
    {
        var interval = publishingInterval > TimeSpan.Zero ? publishingInterval : _options.DefaultPollInterval;
        return Task.FromResult(_poll.Subscribe(fullReferences, interval));
    }

    /// <inheritdoc />
    public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
    {
        _poll.Unsubscribe(handle);
        return Task.CompletedTask;
    }

    // ---- IHostConnectivityProbe ----

    /// <summary>
    ///     The single logical host this instance dials. There is no background probe loop: the state is a
    ///     by-product of the Initialize-time liveness check and of every poll, so the report is always a
    ///     statement about traffic that actually happened rather than about a synthetic ping.
    /// </summary>
    /// <returns>A one-element list describing the configured database endpoint.</returns>
    public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
    {
        lock (_probeLock)
            return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
    }

    // ---- health + host-state classification ----

    /// <summary>
    ///     Classifies one poll's snapshots into a health verdict and a host state.
    ///     <list type="bullet">
    ///         <item>Any Good snapshot ⇒ the database answered: <c>LastSuccessfulRead</c> advances and the
    ///         host is Running.</item>
    ///         <item>Any <b>connection-class</b> Bad snapshot (<see cref="SqlStatusCodes.BadTimeout"/> /
    ///         <see cref="SqlStatusCodes.BadCommunicationError"/>) ⇒ Degraded; when nothing at all was Good,
    ///         the host is Stopped.</item>
    ///         <item>Bad codes that are <b>authoring</b> facts — an unresolvable RawPath, an absent row, a
    ///         type mismatch, a rejected definition — change nothing. A tag typo must never report the
    ///         database as down and send an operator to the wrong system.</item>
    ///     </list>
    ///     <para><b>Why this does not throw to force poll-engine backoff.</b> Backoff would require turning
    ///     an all-<c>BadTimeout</c> batch into an exception, and the engine's exception path publishes
    ///     nothing — clients would lose the very BadTimeout quality the reader went out of its way to
    ///     produce, and would sit on stale Good values instead. Nor is hammering a real risk: each group is
    ///     already bounded by <c>operationTimeout</c> and the reader never holds more than
    ///     <c>maxConcurrentGroups</c> connections however long the database stays frozen. So a frozen
    ///     database is reported (Degraded + host Stopped + Bad-quality values) at the configured poll
    ///     cadence rather than backed off from.</para>
    /// </summary>
    /// <param name="snapshots">The snapshots the reader returned for one poll.</param>
    internal void ObservePollOutcome(IReadOnlyList<DataValueSnapshot> snapshots)
    {
        if (snapshots.Count == 0) return;

        var good = 0;
        var unreachable = 0;
        foreach (var snapshot in snapshots)
        {
            if (SqlStatusCodes.IsGood(snapshot.StatusCode)) good++;
            else if (IsConnectionClass(snapshot.StatusCode)) unreachable++;
        }

        if (unreachable > 0)
        {
            Degrade(
                $"{unreachable} of {snapshots.Count} Sql tag(s) timed out or failed communication against " +
                $"{Endpoint} on the last poll.");
            if (good > 0)
            {
                // Partial: something answered, so the endpoint is up — but the driver is not healthy.
                TouchLastSuccessfulRead();
                TransitionHostTo(HostState.Running);
            }
            else
            {
                TransitionHostTo(HostState.Stopped);
            }

            return;
        }

        if (good == 0) return;   // authoring-class Bad only — not a statement about the database.

        // Route through the Faulted guard, not a raw WriteHealth: a poll still in flight when a concurrent
        // ReinitializeAsync has recorded Faulted (the engine's ~5 s teardown wait is shorter than the 15 s
        // default operationTimeout, so a stale poll can still complete afterwards) must not flip Faulted back
        // to Healthy with no real recovery. Only a successful (re)initialize clears Faulted.
        SetHealthUnlessFaulted(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
        TransitionHostTo(HostState.Running);
    }

    /// <summary>True for the status codes that mean "the database did not answer", as opposed to "the data is not there".</summary>
    private static bool IsConnectionClass(uint statusCode)
        => statusCode is SqlStatusCodes.BadTimeout or SqlStatusCodes.BadCommunicationError;

    /// <summary>
    ///     Routes a poll-loop reader failure to the health surface (05/STAB-9). Deliberately does <b>not</b>
    ///     touch host state: the engine also reports its own contract violations here, which say nothing
    ///     about connectivity — the unreachable-database transition is raised by <see cref="ReadAsync"/>,
    ///     which knows the exception came from the reader.
    /// </summary>
    /// <param name="ex">The exception the poll engine caught.</param>
    internal void HandlePollError(Exception ex)
    {
        // A subscribed-read outage is classified by ReadAsync — good, endpoint-bearing LastError + host
        // Stopped — before it rethrows, and the poll engine then routes that SAME exception object here.
        // Re-degrading would overwrite the good LastError with a worse, guidance-free one and log the outage
        // a second time, so defer to the classification that already happened.
        //
        // Matched on the exception INSTANCE ReadAsync marked, not on its type. `ex is DbException` was the
        // stated intent — "the class ReadAsync owns" — but not what it implemented (Driver.Sql-004): the
        // reader raises a plain InvalidOperationException when the provider factory returns no connection,
        // and a provider can raise ArgumentException/FormatException off a malformed connection string on
        // the same open path. Those are classified by ReadAsync exactly like a DbException, yet a type test
        // let them through and clobbered its message; conversely a DbException that never went through
        // ReadAsync — the poll engine's own throw — was swallowed. Consumed once, so a later unclassified
        // exception is never mistaken for this one.
        if (ReferenceEquals(Interlocked.CompareExchange(ref _classifiedByRead, null, ex), ex)) return;

        // Whatever reaches here — an engine contract-violation, or a non-DbException the provider raised
        // (e.g. an ArgumentException from a malformed connection string, which the parser echoes with
        // credential fragments) — the operator surface carries only the exception TYPE, never ex.Message.
        // The full exception, with its text, still reaches the log SINK via the exception parameter. This is
        // the same unconditional rule InitializeAsync/ReadAsync follow; do not weaken it to a type check.
        _logger.LogWarning(ex, "Sql poll failed. Driver={DriverInstanceId} Endpoint={Endpoint}",
            _driverInstanceId, Endpoint);
        Degrade($"Sql driver '{_driverInstanceId}' poll failed: {ex.GetType().Name}.");
    }

    /// <summary>
    ///     Degrades health, preserving <c>LastSuccessfulRead</c> and never downgrading
    ///     <see cref="DriverState.Faulted"/> — a config-level verdict that only a successful read or an
    ///     operator reinitialize may clear.
    /// </summary>
    /// <param name="reason">The operator-facing reason, which must never carry the connection string.</param>
    private void Degrade(string reason)
    {
        var current = ReadHealth();
        SetHealthUnlessFaulted(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, reason));
    }

    /// <summary>
    ///     Publishes <paramref name="value"/> unless the driver is already <see cref="DriverState.Faulted"/>
    ///     — a liveness/config verdict that only a successful (re)initialize may clear. Every non-Initialize
    ///     health write routes through here: <see cref="Degrade"/> and the poll's Healthy verdict in
    ///     <see cref="ObservePollOutcome"/>, so a late in-flight poll cannot un-fault a driver a concurrent
    ///     <see cref="ReinitializeAsync"/> just faulted.
    /// </summary>
    /// <param name="value">The health snapshot to publish when the driver is not Faulted.</param>
    private void SetHealthUnlessFaulted(DriverHealth value)
    {
        if (ReadHealth().State == DriverState.Faulted) return;
        WriteHealth(value);
    }

    /// <summary>Advances <c>LastSuccessfulRead</c> without changing the current state or error.</summary>
    private void TouchLastSuccessfulRead()
    {
        var current = ReadHealth();
        WriteHealth(current with { LastSuccessfulRead = DateTime.UtcNow });
    }

    /// <summary>Barrier-protected read of the multi-threaded health field.</summary>
    private DriverHealth ReadHealth() => Volatile.Read(ref _health);

    /// <summary>Barrier-protected publish of a new health snapshot.</summary>
    private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);

    /// <summary>Records a host-state change and raises <see cref="OnHostStatusChanged"/> — on transitions only.</summary>
    private void TransitionHostTo(HostState newState)
    {
        HostState old;
        lock (_probeLock)
        {
            old = _hostState;
            if (old == newState) return;
            _hostState = newState;
            _hostStateChangedUtc = DateTime.UtcNow;
        }

        OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, old, newState));
    }

    // ---- authored tag table ----

    /// <summary>
    ///     Maps every authored <see cref="RawTagEntry"/> through <see cref="SqlEquipmentTagParser.TryParse"/>
    ///     and publishes the result as one atomic snapshot. A blob that does not map is <b>logged and
    ///     skipped</b>, never thrown: one malformed tag must not take the whole driver — and therefore every
    ///     other tag on that database — down with it. The skipped RawPath then resolves to nothing, so it
    ///     publishes <see cref="SqlStatusCodes.BadNodeIdUnknown"/> rather than someone else's row.
    /// </summary>
    private void BuildTagTable()
    {
        var table = new Dictionary<string, SqlTagDefinition>(_options.RawTags.Count, StringComparer.Ordinal);
        var blobSkips = new List<SkippedTag>();
        foreach (var entry in _options.RawTags)
        {
            try
            {
                if (SqlEquipmentTagParser.TryParse(entry.TagConfig, entry.RawPath, out var definition))
                    table[entry.RawPath] = definition;
                else
                {
                    blobSkips.Add(new SkippedTag(entry.RawPath, "TagConfig did not map to a definition"));
                    _logger.LogWarning(
                        "Sql tag config did not map to a definition; skipping. Driver={DriverInstanceId} RawPath={RawPath}",
                        _driverInstanceId, entry.RawPath);
                }
            }
            catch (Exception ex)
            {
                // TryParse is contracted never to throw, but BuildTagTable runs BEFORE InitializeAsync's
                // try/catch, so a stray throw here (a parser bug on one blob) would escape Initialize and
                // strand health at Initializing across every DriverInstanceActor retry. Skip the offending
                // tag — the same "one bad tag must not take the whole driver down" rule the malformed-blob
                // branch above follows — rather than fault the driver over a single entry.
                blobSkips.Add(new SkippedTag(entry.RawPath, $"TagConfig threw while parsing: {ex.GetType().Name}"));
                _logger.LogError(
                    ex, "Sql tag config threw while parsing; skipping. Driver={DriverInstanceId} RawPath={RawPath}",
                    _driverInstanceId, entry.RawPath);
            }
        }

        // The gate has not run yet, so the report covers blob-level skips only; ApplyCatalogGateAsync
        // republishes it with its own rejections folded in. Publishing here rather than waiting means a
        // driver whose gate never runs (nothing authored, or the gate threw) still reports what it knows.
        _blobSkips = blobSkips;
        PublishSkipLedger([]);

        Volatile.Write(ref _tagsByRawPath, table);

        // Nothing is pollable until the catalog gate has validated it. Clearing here (rather than leaving a
        // previous generation's validated table in place) means a Reinitialize whose gate then fails cannot
        // keep polling the OLD identifiers against a database whose schema may be exactly what changed.
        Volatile.Write(ref _polledByRawPath, new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal));
    }

    // ---- catalog gate (design §8.1) ----

    /// <summary>
    ///     Validates every authored identifier against the live catalog and republishes the tag table with
    ///     catalog spellings, dropping the tags that do not resolve (design §8.1).
    /// </summary>
    /// <remarks>
    ///     <para><b>Why this must run before the driver reports Healthy.</b> The table this swaps in is what
    ///     every poll resolves against. Running it later — or in the background — would leave a window in
    ///     which an unvalidated identifier could be quoted into a query, which is the entire hole the gate
    ///     exists to close.</para>
    ///     <para><b>A dropped tag is not a driver fault.</b> It disappears from the table, so its RawPath
    ///     resolves to nothing and it publishes <see cref="SqlStatusCodes.BadNodeIdUnknown"/> — one operator
    ///     typo must not stop the other tags on that database, the same rule
    ///     <see cref="BuildTagTable"/> follows for a malformed blob. Each drop is logged at Warning with the
    ///     tag, the field and the reason, because a node that silently goes Bad with nothing in the log is
    ///     unsupportable.</para>
    ///     <para><b>No tags ⇒ no I/O.</b> A driver with nothing authored has nothing to validate, and issuing
    ///     catalog queries to prove that would be a round-trip that can only fail.</para>
    ///     <para>Bounded by the same wall-clock discipline as the liveness check (R2-01 / STAB-14): the work
    ///     runs on the thread pool so a provider implementing the async path synchronously blocks a pool
    ///     thread rather than wedging Initialize forever with no retry.</para>
    /// </remarks>
    /// <param name="cancellationToken">The caller's token.</param>
    /// <returns>A task that completes when the tag table has been validated and republished.</returns>
    private async Task ApplyCatalogGateAsync(CancellationToken cancellationToken)
    {
        var authored = Tags;
        if (authored.Count == 0) return;

        var tables = authored.Values
            .Select(definition => definition.Table)
            .Where(table => !string.IsNullOrWhiteSpace(table))
            .Distinct(StringComparer.OrdinalIgnoreCase)
            .ToArray();

        var catalog = await RunBoundedAsync(
            token => LoadCatalogAsync(tables, token),
            _options.OperationTimeout,
            "the catalog queries did not return",
            cancellationToken).ConfigureAwait(false);

        var result = SqlCatalogGate.Apply(authored.Values, catalog, _dialect);

        // Gitea #541: a gate rejection leaves the tag unpolled and publishing BadNodeIdUnknown with the
        // driver Healthy — the same invisible outcome as a blob that did not map, so it is the same signal.
        PublishSkipLedger(result.Rejected.Select(r =>
            new SkippedTag(r.RawPath, $"rejected by the catalog gate: {r.Field} — {r.Reason}")));

        foreach (var rejection in result.Rejected)
        {
            _logger.LogWarning(
                "Sql tag '{Tag}' rejected by the catalog gate: {Field} — {Reason}. It will publish "
                + "BadNodeIdUnknown until the tag or the database is corrected. Driver={DriverInstanceId}",
                rejection.RawPath, rejection.Field, rejection.Reason, _driverInstanceId);
        }

        var validated = new Dictionary<string, SqlTagDefinition>(result.Accepted.Count, StringComparer.Ordinal);
        foreach (var definition in result.Accepted) validated[definition.Name] = definition;

        // Only the POLLED table is replaced. The authored table is left intact so every authored tag keeps
        // its node and a rejected one publishes BadNodeIdUnknown rather than vanishing (see _polledByRawPath).
        Volatile.Write(ref _polledByRawPath, validated);

        _logger.LogInformation(
            "Sql driver {DriverInstanceId} catalog gate accepted {Accepted} of {Total} authored tag(s) "
            + "across {Tables} table(s) on {Endpoint}.",
            _driverInstanceId, result.Accepted.Count, authored.Count, tables.Length, Endpoint);
    }

    /// <summary>Opens one connection and reads the catalog slice the authored tables need.</summary>
    private async Task<SqlCatalog> LoadCatalogAsync(
        IReadOnlyCollection<string> authoredTables, CancellationToken cancellationToken)
    {
        var connection = _factory.CreateConnection()
                         ?? throw new InvalidOperationException(
                             $"the {_dialect.Provider} provider factory returned no connection.");

        await using (connection.ConfigureAwait(false))
        {
            connection.ConnectionString = _connectionString;
            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
            return await SqlCatalogLoader
                .LoadAsync(connection, _dialect, authoredTables, _options.CommandTimeout, cancellationToken)
                .ConfigureAwait(false);
        }
    }

    // ---- liveness ----

    /// <summary>
    ///     Opens ONE connection, runs the dialect's liveness statement, and disposes it. No connection is
    ///     held between this check and the first poll — the reader opens, uses and disposes per poll, so a
    ///     long-lived connection here would only be a second thing to keep alive.
    ///     <para><b>Bounded by wall-clock, not only by the token</b> (the R2-01 / STAB-14 lesson): some ADO.NET
    ///     providers implement the async path synchronously, and a wedged socket can hang inside the
    ///     provider's own cancellation handshake. If Initialize hung there, <c>DriverInstanceActor</c>'s init
    ///     task would never complete and the driver would sit in Connecting forever with no retry. The work
    ///     runs on the thread pool so a synchronous provider blocks a pool thread rather than the caller, and
    ///     an abandoned attempt still owns and disposes its own connection.</para>
    /// </summary>
    /// <param name="cancellationToken">The caller's token.</param>
    /// <returns>A task that completes when the database has answered.</returns>
    private Task VerifyLivenessAsync(CancellationToken cancellationToken) =>
        RunBoundedAsync(
            async token => { await PingAsync(token).ConfigureAwait(false); return true; },
            _options.CommandTimeout,
            "the liveness statement did not return",
            cancellationToken);

    /// <summary>
    ///     Runs <paramref name="work"/> under a hard wall-clock <paramref name="budget"/>, on the thread pool,
    ///     converting a breach into a <see cref="TimeoutException"/> worded from <paramref name="what"/>.
    /// </summary>
    /// <remarks>
    ///     <para>The R2-01 / STAB-14 shape, shared by the two pieces of I/O Initialize performs. It exists
    ///     because a token is not a deadline: some ADO.NET providers implement the async path synchronously,
    ///     and a wedged socket can hang <em>inside</em> the provider's own cancellation handshake. Without an
    ///     independent wall clock, <c>DriverInstanceActor</c>'s init task would never complete and the driver
    ///     would sit in Connecting forever with no retry — the exact frozen-peer wedge the S7 driver shipped.</para>
    ///     <para>Both the linked CTS deadline and an outer <see cref="TaskExtensions"/> wait are used, because
    ///     each covers a case the other does not: the CTS handles a provider that <em>does</em> honour
    ///     cancellation, and the outer wait handles one that does not. An abandoned attempt keeps ownership of
    ///     its own connection and disposes it; <see cref="Detach"/> observes its eventual fault.</para>
    /// </remarks>
    /// <typeparam name="T">The work's result type.</typeparam>
    /// <param name="work">The operation to bound. Receives the deadline-linked token.</param>
    /// <param name="budget">The wall-clock allowance.</param>
    /// <param name="what">Sentence fragment naming the operation, e.g. "the catalog queries did not return".</param>
    /// <param name="cancellationToken">The caller's token.</param>
    /// <returns>The work's result.</returns>
    /// <exception cref="TimeoutException">The budget elapsed first.</exception>
    private static async Task<T> RunBoundedAsync<T>(
        Func<CancellationToken, Task<T>> work,
        TimeSpan budget,
        string what,
        CancellationToken cancellationToken)
    {
        var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        Task<T> running;
        try
        {
            deadline.CancelAfter(budget);
            running = Task.Run(
                async () =>
                {
                    try { return await work(deadline.Token).ConfigureAwait(false); }
                    finally { deadline.Dispose(); }
                },
                CancellationToken.None);
        }
        catch
        {
            // Ownership of the CTS transfers to the work task; release it only if that task never started.
            deadline.Dispose();
            throw;
        }

        try
        {
            return await running.WaitAsync(budget, cancellationToken).ConfigureAwait(false);
        }
        catch (TimeoutException)
        {
            Detach(running);
            throw new TimeoutException($"{what} within {(int)budget.TotalMilliseconds} ms.");
        }
        catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
        {
            // Our deadline fired and the provider DID honour the linked token.
            Detach(running);
            throw new TimeoutException($"{what} within {(int)budget.TotalMilliseconds} ms.");
        }
    }

    /// <summary>Opens a connection and executes the dialect's cheapest proof-of-life statement.</summary>
    private async Task PingAsync(CancellationToken cancellationToken)
    {
        var connection = _factory.CreateConnection()
                         ?? throw new InvalidOperationException(
                             $"the {_dialect.Provider} provider factory returned no connection.");

        await using (connection.ConfigureAwait(false))
        {
            connection.ConnectionString = _connectionString;
            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);

            var command = connection.CreateCommand();
            await using (command.ConfigureAwait(false))
            {
                command.CommandText = _dialect.LivenessSql;
                // ADO.NET reads CommandTimeout = 0 as "wait forever" — the one value it must never become.
                command.CommandTimeout = Math.Max(1, (int)Math.Ceiling(_options.CommandTimeout.TotalSeconds));
                await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
            }
        }
    }

    /// <summary>
    ///     Observes an abandoned liveness attempt's eventual fault so it never surfaces as an unobserved
    ///     task exception. The task still owns — and disposes — its own connection.
    ///     <para>TODO(M2): a verbatim duplicate of <c>SqlPollReader.Detach</c>; both live in Driver.Sql and
    ///     could hoist to one internal helper. Left in place here to keep this change off the reader.</para>
    /// </summary>
    private static void Detach(Task work)
        => _ = work.ContinueWith(
            static t => _ = t.Exception,
            CancellationToken.None,
            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
            TaskScheduler.Default);

    // ---- endpoint description ----

    /// <summary>
    ///     Renders a connection string as <c>server/database</c>, reading only those two keys.
    ///     <para>This is the <b>only</b> function permitted to look at the connection string outside the
    ///     provider call, and it exists so that no other code is ever tempted to log the string itself:
    ///     credentials live in <c>User ID</c> / <c>Password</c> / <c>Authentication</c>, which are never
    ///     read here. A string that cannot be parsed yields <see cref="UnknownEndpoint"/> — an unusable
    ///     description must not become a crash at construction.</para>
    /// </summary>
    /// <param name="connectionString">The resolved connection string.</param>
    /// <returns>A credential-free description safe to log and display.</returns>
    private static string DescribeEndpoint(string connectionString)
    {
        try
        {
            var builder = new DbConnectionStringBuilder { ConnectionString = connectionString };
            var server = FirstValue(builder, ServerKeys);
            if (string.IsNullOrWhiteSpace(server)) return UnknownEndpoint;
            var database = FirstValue(builder, DatabaseKeys);
            return string.IsNullOrWhiteSpace(database) ? server : $"{server}/{database}";
        }
        catch (ArgumentException)
        {
            return UnknownEndpoint;
        }
    }

    /// <summary>Reads the first of <paramref name="keys"/> the builder carries; null when it carries none.</summary>
    private static string? FirstValue(DbConnectionStringBuilder builder, string[] keys)
    {
        foreach (var key in keys)
        {
            // DbConnectionStringBuilder's key comparison is case-insensitive.
            if (builder.TryGetValue(key, out var value) && value is not null)
            {
                var text = value.ToString();
                if (!string.IsNullOrWhiteSpace(text)) return text;
            }
        }

        return null;
    }

    // ---- teardown ----

    /// <summary>
    ///     Performs the same teardown as <see cref="ShutdownAsync"/>, so a caller that only uses
    ///     <c>await using</c> does not leak the poll loops.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync() => await TeardownAsync().ConfigureAwait(false);

    /// <summary>
    ///     Shared teardown: stops every polled subscription and drops the authored table. Idempotent — safe
    ///     to call any number of times, in any order, which is what makes
    ///     <c>ShutdownAsync</c>-then-<c>DisposeAsync</c> (and <see cref="ReinitializeAsync"/>) safe.
    ///     <para>There is no connection to close: the reader opens and disposes one per poll.</para>
    /// </summary>
    private async Task TeardownAsync()
    {
        await _poll.DisposeAsync().ConfigureAwait(false);
        Volatile.Write(ref _tagsByRawPath, new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal));
        Volatile.Write(ref _polledByRawPath, new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal));
        _resolver.Clear();   // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
    }
}