lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs

using System.Collections.Concurrent;
using System.Collections.Frozen;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;

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

/// <summary>
///     The MQTT / Sparkplug B driver shell: composes <see cref="MqttConnection"/> (broker session +
///     reconnect supervisor), <see cref="MqttSubscriptionManager"/> (authored table, topic routing,
///     value extraction) and the manager's <see cref="LastValueCache"/> (the push→poll bridge) into
///     the driver capability set the Core consumes.
/// </summary>
/// <remarks>
///     <para>
///         <b>Authored-only discovery.</b> Plain MQTT has no browsable address space — a broker will
///         happily carry thousands of topics this deployment has nothing to do with.
///         <see cref="DiscoverAsync"/> therefore replays exactly the raw tags the deploy artifact
///         authored (<see cref="MqttDriverOptions.RawTags"/>) and nothing else, and
///         <see cref="ITagDiscovery.SupportsOnlineDiscovery"/> stays <see langword="false"/> so the
///         universal browser never treats broker traffic as a tag catalogue. In
///         <see cref="MqttMode.Plain"/> the policy is <see cref="DiscoveryRediscoverPolicy.Once"/>:
///         the authored set — and every authored tag's datatype — is fully known synchronously, so
///         re-running discovery on a timer can only produce the same tree.
///     </para>
///     <para>
///         <b>Two ingest paths, one driver.</b> <see cref="MqttMode.Plain"/> routes through
///         <see cref="MqttSubscriptionManager"/> (authored topics); <see cref="MqttMode.SparkplugB"/>
///         routes through <see cref="SparkplugIngestor"/> (one group-wide filter, birth certificates,
///         alias resolution, seq gaps, rebirth NCMDs). Exactly one is live, selected by
///         <see cref="MqttDriverOptions.Mode"/> and rebuilt whenever
///         <see cref="ReinitializeAsync"/> changes anything either captures. They share the
///         driver-owned <see cref="LastValueCache"/>, so <c>IReadable</c> reads one store either way.
///     </para>
///     <para>
///         <b><see cref="OnRediscoveryNeeded"/> fires in Sparkplug B only, and only on a real change.</b>
///         A Sparkplug tag's <c>dataType</c> is optional — the birth certificate declares it — so the
///         discovered tree fills in asynchronously and the policy is
///         <see cref="DiscoveryRediscoverPolicy.UntilStable"/>. <see cref="SparkplugIngestor.BirthObserved"/>
///         is wired to <see cref="RaiseRediscoveryNeeded"/> through
///         <see cref="OnBirthObserved"/>, which fires <b>only</b> when the birth's metric-name SET
///         differs from the last one seen for that scope (and only for scopes an authored tag actually
///         binds). That gate is the anti-storm mechanism, not an optimisation: rediscovery triggers an
///         address-space rebuild, edge nodes re-birth freely — on their own timer, on every reconnect
///         via the late-join rebirth, and once per NCMD the ingestor's gap policy sends — and firing on
///         each of those would turn a healthy plant into a permanent rebuild loop. Plain mode never
///         raises it at all; its authored set changes only by redeploy.
///     </para>
///     <para>
///         <b>Composition order is load-bearing.</b> <see cref="MqttSubscriptionManager.AttachTo"/>
///         must run <i>before</i> any subscribe and before the connect completes — it is what wires
///         message delivery and, critically, the reconnect re-subscribe. The connection's
///         <c>Reconnected</c> handler is passed through unwrapped: the manager throws when it can
///         re-establish nothing, and that throw is precisely what tears the session down and retries.
///         Swallowing it would leave a driver that reports Connected and receives nothing.
///     </para>
///     <para>
///         <b>Constructor is connection-free.</b> It maps the authored TagConfig blobs (pure, no I/O)
///         so discovery and reads work off a configured-but-not-yet-connected driver; every network
///         operation happens in <see cref="InitializeAsync"/>.
///     </para>
/// </remarks>
public sealed class MqttDriver
    : IDriver, ITagDiscovery, ISubscribable, IReadable, IWritable, IHostConnectivityProbe, IRediscoverable,
      ISkippedTagReporter, IAsyncDisposable, IDisposable
{
    /// <summary>Authored tags this driver could not build into its ingest table (Gitea #541).</summary>
    private readonly SkippedTagLedger _skippedTags = new();

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

    // ---- Write status codes (Gitea #508) ----
    //
    // NAMED constants, never inline literals: StatusCodeParityTests checks each against the pinned SDK
    // by reflection and can only see a named const (Gitea #497/#519). The names must match
    // Opc.Ua.StatusCodes members exactly.

    /// <summary>The reference is not an authored tag of this driver's live ingest path.</summary>
    private const uint StatusBadNodeIdUnknown = 0x80340000u;

    /// <summary>
    ///     The tag declares no write destination — a Plain tag with no <c>writeTopic</c>, or a Sparkplug
    ///     tag with an incomplete binding tuple. The honest answer to "can I write this?", and the whole
    ///     reason the encoder refuses rather than no-opping.
    /// </summary>
    private const uint StatusBadNotWritable = 0x803B0000u;

    /// <summary>The supplied value cannot be represented in the tag's declared data type.</summary>
    private const uint StatusBadTypeMismatch = 0x80740000u;

    /// <summary>The publish failed, or the driver has no live broker connection to publish through.</summary>
    private const uint StatusBadCommunicationError = 0x80050000u;

    /// <summary>
    ///     The single folder <see cref="DiscoverAsync"/> streams every variable under — and therefore
    ///     the only subtree path a rediscovery <c>ScopeHint</c> can name truthfully. See
    ///     <see cref="OnBirthObserved"/>.
    /// </summary>
    private const string DiscoveryFolderName = "Mqtt";

    /// <summary>How often the host-connectivity poll samples <see cref="MqttConnection.State"/>.</summary>
    private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);

    /// <summary>
    ///     Stand-in for a non-positive authored <c>connectTimeoutSeconds</c> — see <see cref="Normalize"/>.
    ///     Matches <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/>'s own default.
    /// </summary>
    private const int DefaultConnectTimeoutSeconds = 15;

    /// <summary>
    ///     Stand-in for a negative authored <c>keepAliveSeconds</c> — see <see cref="Normalize"/>.
    ///     Matches <see cref="MqttDriverOptions.KeepAliveSeconds"/>'s own default.
    /// </summary>
    private const int DefaultKeepAliveSeconds = 30;

    private readonly string _driverInstanceId;
    private readonly ILogger? _logger;

    /// <summary>
    ///     Owned by the driver, not by the manager, so a manager rebuilt for changed ingest settings
    ///     inherits the observed values instead of silently regressing every node to
    ///     <c>BadWaitingForInitialData</c>.
    /// </summary>
    private readonly LastValueCache _values = new();

    private readonly object _hostLock = new();

    /// <summary>
    ///     Per authored Sparkplug scope, the signature of the metric-name set its most recent birth
    ///     declared — the rediscovery change gate's memory. Written on MQTTnet's dispatcher thread and
    ///     read nowhere else, but concurrent because a reconnect's birth flood and the dispatcher are
    ///     not guaranteed to be the same thread forever.
    /// </summary>
    /// <remarks>
    ///     <b>Deliberately NOT cleared on a death, a reconnect or an ingest rebuild.</b> The ingestor
    ///     drops its whole <see cref="SparkplugIngestor.Births"/> cache on every reconnect (an alias may
    ///     have been rebound while the driver was away) — but "the driver forgot" is not "the address
    ///     space changed". Clearing this alongside it would make every reconnect's late-join rebirth
    ///     flood look like a tree full of brand-new scopes and fire one rediscovery per authored scope
    ///     on every single flap, which is the storm this map exists to prevent. It survives so that an
    ///     identical rebirth stays silent.
    /// </remarks>
    private readonly ConcurrentDictionary<SparkplugScope, string> _birthSignatures = new();

    /// <summary>Guards <see cref="InitializeAsync"/> / <see cref="ReinitializeAsync"/> / <see cref="ShutdownAsync"/> against each other.</summary>
    private readonly SemaphoreSlim _lifecycleGate = new(1, 1);

    // _options / _subscriptions / _authoredRawPaths are written only under _lifecycleGate (or in the
    // ctor) and read freely elsewhere, WITHOUT the Volatile discipline _health / _hostState get.
    // That is deliberate, not an oversight: all three are reference/interface fields, so a reader
    // sees either the whole old object or the whole new one — never a torn one. The looser fields
    // are the ones a reader may legitimately observe one publish stale (a read served from the
    // previous authored table is a correct read of a value that was correct a microsecond ago);
    // _health and _hostState feed ServiceLevel and the status dashboard, where a stale per-core copy
    // is an operator-visible lie about whether the driver is up.
    private MqttDriverOptions _options;
    private MqttSubscriptionManager _subscriptions;

    /// <summary>
    ///     The Sparkplug B ingest path, or <see langword="null"/> in <see cref="MqttMode.Plain"/>.
    ///     Exactly one of this and <see cref="_subscriptions"/> is fed authored tags and attached to the
    ///     connection; the other exists but is never registered against, so it can neither route a
    ///     message nor answer a resolve.
    /// </summary>
    private SparkplugIngestor? _sparkplug;

    /// <summary>
    ///     RawPaths of the currently registered authored tags, in authoring order — the discovery
    ///     enumeration set. The manager resolves a RawPath to its definition but does not enumerate,
    ///     and enumerating from here is what makes "authored only" structural rather than incidental.
    /// </summary>
    private IReadOnlyList<string> _authoredRawPaths = [];

    /// <summary>
    ///     The Sparkplug scopes at least one authored tag binds — empty outside
    ///     <see cref="MqttMode.SparkplugB"/>. Rebuilt wholesale by <see cref="RegisterAuthoredTags"/>
    ///     alongside the authored table it is derived from.
    /// </summary>
    /// <remarks>
    ///     The rediscovery gate's first question. The ingestor applies a birth for <i>any</i> device
    ///     under an authored edge node (it has to — a DBIRTH is a sequenced member of that node's
    ///     stream), so on a large plant it observes births for devices this deployment never reads. Such
    ///     a birth cannot change <see cref="DiscoverAsync"/>'s output by construction, because discovery
    ///     replays the authored set — so firing rediscovery for it would rebuild the address space in
    ///     response to traffic the configuration has no opinion about. Gating on this set also bounds
    ///     <see cref="_birthSignatures"/> by <i>configuration</i> rather than by whatever the plant
    ///     happens to publish — the same rule the ingestor's own authored-edge-node filter follows.
    /// </remarks>
    private FrozenSet<SparkplugScope> _authoredScopes = FrozenSet<SparkplugScope>.Empty;

    private MqttConnection? _connection;
    private CancellationTokenSource? _hostProbeCts;

    /// <summary>The Primary data-plane verdict the Sparkplug host claim is gated on (Gitea #511).</summary>
    private readonly IDriverPrimaryRoleSource _primaryRoleSource;

    /// <summary>
    ///     The Sparkplug primary-host claim, or <see langword="null"/> when this configuration does not
    ///     ask for one. Owns its own broker connection — see <see cref="SparkplugPrimaryHost"/> for why
    ///     it cannot share the ingest session.
    /// </summary>
    private SparkplugPrimaryHost? _primaryHost;

    private DriverHealth _health = new(DriverState.Unknown, null, null);
    private HostState _hostState = HostState.Unknown;
    private DateTime _hostStateChangedUtc = DateTime.UtcNow;
    private int _disposed;

    /// <summary>Initializes a new driver. Stores + maps configuration only — no network, no client.</summary>
    /// <param name="options">
    ///     Driver configuration, including the authored <see cref="MqttDriverOptions.RawTags"/>.
    ///     Task 9's factory deserializes it from the <c>DriverConfig</c> JSON; a later
    ///     <see cref="ReinitializeAsync"/> may replace it from a fresh config blob.
    /// </param>
    /// <param name="driverInstanceId">Stable logical id of this driver instance.</param>
    /// <param name="logger">Optional logger; never receives credentials or payload bodies.</param>
    /// <param name="primaryRoleSource">
    ///     Gitea #511: the Primary data-plane verdict the Sparkplug primary-host claim is gated on.
    ///     <see langword="null"/> falls back to <see cref="UnmanagedDriverPrimaryRoleSource"/> — which
    ///     always claims — for standalone hosts and tests. On a redundant pair that fallback is the
    ///     dual-claim this gate exists to prevent, so the Host resolves a real one and
    ///     <c>DriverFactoryBootstrap</c> does so with <c>GetRequiredService</c>.
    /// </param>
    public MqttDriver(
        MqttDriverOptions options,
        string driverInstanceId,
        ILogger<MqttDriver>? logger = null,
        IDriverPrimaryRoleSource? primaryRoleSource = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);

        _options = Normalize(options);
        _driverInstanceId = driverInstanceId;
        _logger = logger;
        _primaryRoleSource = primaryRoleSource ?? UnmanagedDriverPrimaryRoleSource.Instance;
        _subscriptions = CreateSubscriptionManager(_options);
        _sparkplug = CreateSparkplugIngestor(_options);
        RegisterAuthoredTags(_options);
    }

    /// <inheritdoc />
    public event EventHandler<DataChangeEventArgs>? OnDataChange;

    /// <inheritdoc />
    public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;

    /// <inheritdoc />
    public event EventHandler<RediscoveryEventArgs>? OnRediscoveryNeeded;

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

    /// <inheritdoc />
    public string DriverType => DriverTypeNames.Mqtt;

    /// <summary>
    ///     The ingest path this driver composes. Internal: the P2 Sparkplug handler feeds the same
    ///     <c>OnDataChange</c> + <see cref="LastValueCache"/> sinks through it, and the shell tests
    ///     drive <see cref="MqttSubscriptionManager.HandleMessage"/> to simulate broker traffic
    ///     without a broker.
    /// </summary>
    internal MqttSubscriptionManager Subscriptions => _subscriptions;

    /// <summary>
    ///     The Sparkplug B ingest path, or <see langword="null"/> outside
    ///     <see cref="MqttMode.SparkplugB"/>. Internal: the shell tests drive
    ///     <see cref="SparkplugIngestor.Dispatch"/> to simulate an edge node without a broker, and
    ///     Tasks 22/23 read its <see cref="SparkplugIngestor.Births"/>.
    /// </summary>
    internal SparkplugIngestor? Sparkplug => _sparkplug;

    /// <summary>Whether this driver instance ingests Sparkplug B rather than plain topics.</summary>
    private bool IsSparkplug => _options.Mode == MqttMode.SparkplugB;

    /// <summary>
    ///     Test seam: awaited inside <see cref="InitializeCoreAsync"/> <b>while the lifecycle gate is
    ///     held</b>, immediately before the broker connect. Lets a test park a lifecycle operation
    ///     mid-flight and prove <see cref="DisposeAsync"/> serializes behind it rather than racing
    ///     past a half-built session. Mirrors <see cref="MqttConnection"/>'s own
    ///     <c>AfterConnectHookForTests</c>, which exists for the same class of race. Always
    ///     <see langword="null"/> in production.
    /// </summary>
    internal Func<CancellationToken, Task>? BeforeConnectHookForTests { get; set; }

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

    /// <summary>
    ///     Adopts <paramref name="driverConfigJson"/> (falling back to the constructor's options when
    ///     it is blank or unusable), registers the authored tags, and opens the broker session.
    /// </summary>
    /// <remarks>
    ///     Ordering is the contract: register → attach → connect. Attaching before the connect
    ///     completes means the reconnect re-subscribe is wired for the very first drop, and the
    ///     manager's SUBSCRIBE transport is live before the OPC UA server's first subscribe arrives.
    /// </remarks>
    /// <param name="driverConfigJson">The driver configuration as JSON.</param>
    /// <param name="cancellationToken">Cancellation for the connect.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            _lifecycleGate.Release();
        }
    }

    /// <summary>
    ///     Applies a config delta in place. A tag-only delta re-registers the authored table without
    ///     touching the broker session; a delta that changes how the driver connects or ingests
    ///     rebuilds the session.
    /// </summary>
    /// <remarks>
    ///     <b>A bad delta never faults the driver.</b> Unparseable or unusable config is logged and
    ///     discarded, leaving the running configuration exactly as it was — a redeploy carrying one
    ///     malformed driver blob must not take a healthy driver's whole address space Bad. A genuine
    ///     connect failure against a <i>changed</i> endpoint is a different thing and does fault,
    ///     exactly as <see cref="InitializeAsync"/> would.
    /// </remarks>
    /// <param name="driverConfigJson">The driver configuration as JSON.</param>
    /// <param name="cancellationToken">Cancellation for any reconnect the delta forces.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        MqttDriverOptions? next;
        try
        {
            next = ParseOptions(driverConfigJson);
        }
        catch (Exception ex)
        {
            // Defence in depth: ParseOptions already swallows the JSON failure modes.
            _logger?.LogWarning(
                ex,
                "MQTT driver '{DriverId}': reinitialize config could not be read; keeping the running configuration.",
                _driverInstanceId);
            return;
        }

        if (next is null)
        {
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': reinitialize config was blank or unusable; keeping the running configuration.",
                _driverInstanceId);
            return;
        }

        await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            if (!SameSession(next, _options))
            {
                // Broker endpoint / credentials / ingest shape changed — the live session cannot be
                // reused. Full rebuild; a failure here is a real fault, not a bad delta.
                _logger?.LogInformation(
                    "MQTT driver '{DriverId}': reinitialize changes the broker session; rebuilding.",
                    _driverInstanceId);
                await TeardownAsync().ConfigureAwait(false);
                await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
                return;
            }

            _options = next;
            RegisterAuthoredTags(next);
            _logger?.LogInformation(
                "MQTT driver '{DriverId}': reinitialize applied in place; {TagCount} authored tag(s).",
                _driverInstanceId,
                _authoredRawPaths.Count);
        }
        finally
        {
            _lifecycleGate.Release();
        }
    }

    /// <inheritdoc />
    public async Task ShutdownAsync(CancellationToken cancellationToken)
    {
        // GetHealth(), not ReadHealth(): the live connection's LastMessageUtc is the real last-message
        // instant, and the stored snapshot only ever carries what was true at the last initialize. Taking
        // the stored one here throws away the evidence a shutdown-then-initialize cycle (the #527
        // watchdog's reconnect) is measured against, reporting a driver that was delivering a moment ago
        // as one that has never delivered.
        var lastMessage = GetHealth().LastSuccessfulRead;
        if (await _lifecycleGate.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken)
                .ConfigureAwait(false))
        {
            try
            {
                await TeardownAsync().ConfigureAwait(false);
            }
            finally
            {
                _lifecycleGate.Release();
            }
        }
        else
        {
            // A wedged initialize must not turn shutdown into a hang; tear the session down anyway.
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': lifecycle gate still held at shutdown; tearing the session down anyway.",
                _driverInstanceId);
            await TeardownAsync().ConfigureAwait(false);
        }

        WriteHealth(new DriverHealth(DriverState.Unknown, lastMessage, null));
        TransitionHost(HostState.Unknown);
    }

    /// <summary>
    ///     Current health. Derived live from the connection's own state where one exists, so a
    ///     background reconnect is visible without the driver having to mirror every transition.
    ///     <c>LastSuccessfulRead</c> carries the last inbound message instant — MQTT never polls, so
    ///     "last message age" is the only evidence the session is actually delivering.
    /// </summary>
    /// <returns>The driver's current health snapshot.</returns>
    public DriverHealth GetHealth()
    {
        var stored = ReadHealth();
        var connection = _connection;
        if (connection is null)
        {
            return stored;
        }

        var lastMessage = connection.LastMessageUtc ?? stored.LastSuccessfulRead;
        return connection.State switch
        {
            MqttConnectionState.Connected => new DriverHealth(DriverState.Healthy, lastMessage, stored.LastError),
            MqttConnectionState.Reconnecting => new DriverHealth(DriverState.Reconnecting, lastMessage, stored.LastError),
            MqttConnectionState.Faulted => new DriverHealth(DriverState.Faulted, lastMessage, stored.LastError),
            // Disconnected before the first successful connect, or Disposed after teardown: the
            // stored snapshot (Initializing / Faulted / Unknown) is the more informative answer.
            _ => stored with { LastSuccessfulRead = lastMessage },
        };
    }


    // ---- ITagDiscovery ----

    /// <inheritdoc />
    /// <remarks>
    ///     <para>
    ///         Plain mode discovers synchronously from the authored set — the RawPaths <i>and</i> their
    ///         datatypes are both in the deploy artifact — so one pass is all there is.
    ///     </para>
    ///     <para>
    ///         Sparkplug B is <see cref="DiscoveryRediscoverPolicy.UntilStable"/> because half of what
    ///         it discovers arrives on the wire: <c>dataType</c> is optional in the Sparkplug tag shape
    ///         (the birth certificate declares it), so a tag authored with nothing but its
    ///         <c>(group, node, device, metric)</c> tuple can only be streamed with the record's default
    ///         type until its NBIRTH/DBIRTH lands. The host's <c>UntilStable</c> contract — re-run
    ///         discovery until the captured set is non-empty and its signature repeats — is exactly the
    ///         retry that lets those passes catch a birth that had not yet arrived at connect.
    ///     </para>
    ///     <para>
    ///         <b>The retry is a floor, not the mechanism.</b> The host's stability signature is built
    ///         from FullReferences alone, so a pass whose only change is a filled-in datatype still
    ///         looks "stable" and the loop stops. That is why the same fill-in also fires
    ///         <see cref="OnRediscoveryNeeded"/> (see <see cref="OnBirthObserved"/>): a birth arriving
    ///         after the loop settled is the case the timer cannot cover.
    ///     </para>
    /// </remarks>
    public DiscoveryRediscoverPolicy RediscoverPolicy =>
        IsSparkplug ? DiscoveryRediscoverPolicy.UntilStable : DiscoveryRediscoverPolicy.Once;

    /// <inheritdoc />
    /// <remarks>
    ///     Always <see langword="false"/>: <see cref="DiscoverAsync"/> replays authored tags, it does
    ///     not enumerate a backend. A broker's live topic set is not a tag catalogue — the
    ///     <c>Driver.Mqtt.Browser</c> project is the surface for observing traffic during authoring.
    /// </remarks>
    public bool SupportsOnlineDiscovery => false;

    /// <summary>
    ///     Streams the authored tag set — and only that — into <paramref name="builder"/>.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         The enumeration source is the driver's own list of registered RawPaths, never anything
    ///         observed on the wire, so no amount of broker traffic can add a node. A RawPath whose
    ///         TagConfig failed to map is absent from the manager and is skipped here too — it surfaces
    ///         as <c>BadNodeIdUnknown</c> on read rather than as a wrongly-typed variable.
    ///     </para>
    ///     <para>
    ///         <b>Sparkplug B: the datatype is resolved per pass, the tag set never is.</b> An authored
    ///         tag is streamed on <i>every</i> pass, birth or no birth — it is part of the declared
    ///         configuration, not something the plant grants by publishing, and hiding it until a birth
    ///         arrives would make the node's existence depend on wire traffic (and, on a deployment
    ///         whose other tags do carry an authored type, would hide it behind a "non-empty and stable"
    ///         verdict the host reaches without it). What a birth changes is the <i>type</i>: see
    ///         <see cref="ResolveDiscoveredDataType"/>.
    ///     </para>
    /// </remarks>
    /// <param name="builder">The address space builder to stream discovered nodes into.</param>
    /// <param name="cancellationToken">Cancellation for the discovery operation.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(builder);
        cancellationToken.ThrowIfCancellationRequested();

        var folder = builder.Folder(DiscoveryFolderName, DiscoveryFolderName);
        foreach (var rawPath in _authoredRawPaths)
        {
            if (!TryResolve(rawPath, out var def))
            {
                continue;
            }

            folder.Variable(def.Name, def.Name, new DriverAttributeInfo(
                FullName: def.Name,
                DriverDataType: ResolveDiscoveredDataType(def),
                IsArray: false,
                ArrayDim: null,
                // Gitea #508: MQTT now has a write leg, so a tag that declares a write destination is
                // an Operate node. One that does not stays ViewOnly — the point of the distinction is
                // that a node must never look writable while its writes would go nowhere.
                SecurityClass: CanWrite(def) ? SecurityClassification.Operate : SecurityClassification.ViewOnly,
                IsHistorized: false,
                IsAlarm: false,
                // Default-off, deliberately: MQTT gives no application-level write acknowledgement, and
                // a Sparkplug command is not necessarily idempotent (a pulse, a recipe step, a counter
                // increment). The resilience pipeline must not replay one by default; an operator opts
                // in per tag via Tag.WriteIdempotent.
                WriteIdempotent: false));
        }

        return Task.CompletedTask;
    }

    /// <summary>
    ///     The datatype <see cref="DiscoverAsync"/> reports for one authored tag: the authored type when
    ///     the operator declared one, otherwise the type the metric's live birth certificate declared,
    ///     otherwise the record's default.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>The precedence is the ingest path's, restated.</b> <c>SparkplugIngestor.PublishMetric</c>
    ///         coerces every value with <c>DataTypeAuthored ? authored : birth</c>; a discovery surface
    ///         that disagreed with it would advertise a type no value ever arrives as. The two must be
    ///         changed together.
    ///     </para>
    ///     <para>
    ///         An unsupported Sparkplug type (DataSet / Template / PropertySet / Unknown) maps to
    ///         <see langword="null"/> and falls back rather than blanking the tag — the ingest path
    ///         already refuses those metrics once, loudly, and a discovery pass is not the place to
    ///         relitigate it.
    ///     </para>
    /// </remarks>
    /// <param name="def">The authored tag definition.</param>
    /// <returns>The effective data type for this pass.</returns>
    private DriverDataType ResolveDiscoveredDataType(MqttTagDefinition def)
    {
        if (def.DataTypeAuthored || _sparkplug is not { } sparkplug)
        {
            return def.DataType;
        }

        if (def.GroupId is not { } groupId || def.EdgeNodeId is not { } edgeNodeId || def.MetricName is not { } metric)
        {
            return def.DataType;
        }

        var table = sparkplug.Births.Find(new SparkplugScope(groupId, edgeNodeId, def.DeviceId));
        return table?.ResolveByName(metric)?.DataType.ToDriverDataType() ?? def.DataType;
    }

    // ---- ISubscribable (straight through to the manager) ----

    /// <inheritdoc />
    public Task<ISubscriptionHandle> SubscribeAsync(
        IReadOnlyList<string> fullReferences,
        TimeSpan publishingInterval,
        CancellationToken cancellationToken)
        => _sparkplug is { } sparkplug
            ? sparkplug.SubscribeAsync(fullReferences, publishingInterval, cancellationToken)
            : _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken);

    /// <inheritdoc />
    public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
        => _sparkplug is { } sparkplug
            ? sparkplug.UnsubscribeAsync(handle, cancellationToken)
            : _subscriptions.UnsubscribeAsync(handle, cancellationToken);

    /// <summary>Resolves a RawPath through whichever ingest path this driver's mode selected.</summary>
    /// <param name="rawPath">The driver wire reference.</param>
    /// <param name="def">The definition when this returns <see langword="true"/>.</param>
    /// <returns><see langword="true"/> when the reference is an authored tag of the live ingest path.</returns>
    private bool TryResolve(string rawPath, out MqttTagDefinition def)
        => _sparkplug is { } sparkplug
            ? sparkplug.TryResolve(rawPath, out def)
            : _subscriptions.TryResolve(rawPath, out def);

    // ---- IReadable ----

    /// <summary>
    ///     Serves the batch from the last-value cache. MQTT is push-based: there is nothing to poll,
    ///     so a read is always "what was most recently published".
    /// </summary>
    /// <remarks>
    ///     <b>Never throws.</b> A reference that has not been observed — including one that is not an
    ///     authored tag at all — returns <c>BadWaitingForInitialData</c> in its own slot. Failing the
    ///     whole call would let one stale reference blank every other node in the same OPC UA read.
    /// </remarks>
    /// <param name="fullReferences">The RawPath references to read.</param>
    /// <param name="cancellationToken">Unused; the read touches no network.</param>
    /// <returns>One snapshot per requested reference, in request order.</returns>
    public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
        IReadOnlyList<string> fullReferences,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(fullReferences);

        var results = new DataValueSnapshot[fullReferences.Count];
        for (var i = 0; i < fullReferences.Count; i++)
        {
            // The driver-owned cache, not the live ingest path's: both are handed this same instance,
            // and reading it directly means a mid-flight ingest rebuild can never blank a read.
            results[i] = _values.Read(fullReferences[i]);
        }

        return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(results);
    }

    // ---- IWritable (Gitea #508) ----

    /// <summary>
    ///     Whether <paramref name="def"/> declares somewhere for a write to go: a <c>writeTopic</c> in
    ///     Plain mode, or a complete command tuple in Sparkplug mode.
    /// </summary>
    /// <param name="def">The resolved tag definition.</param>
    /// <returns><see langword="true"/> when a write to this tag can be addressed.</returns>
    /// <remarks>
    ///     Asked of the SAME encoder the write path uses, so the browse-time answer and the runtime
    ///     answer cannot drift — a node advertised Operate that then refuses every write would be the
    ///     silent-no-op failure in a new costume. The value is irrelevant here, so a probe value of the
    ///     tag's own type is encoded and thrown away.
    /// </remarks>
    private bool CanWrite(MqttTagDefinition def)
        => MqttWriteEncoder.TryEncode(_options.Mode, def, ProbeValueFor(def.DataType), out _)
           != MqttWriteEncodeError.NoWriteDestination;

    /// <summary>A never-published sentinel value of the given type, for the <see cref="CanWrite"/> probe.</summary>
    /// <param name="dataType">The tag's declared data type.</param>
    /// <returns>A representable value, or a string for types with no natural zero.</returns>
    private static object ProbeValueFor(DriverDataType dataType) => dataType switch
    {
        DriverDataType.Boolean => false,
        DriverDataType.DateTime => DateTime.UnixEpoch,
        DriverDataType.String or DriverDataType.Reference => "",
        _ => 0,
    };

    /// <summary>
    ///     Publishes a batch of writes: one MQTT message per request, to the tag's write topic (Plain)
    ///     or as an NCMD/DCMD command (Sparkplug B).
    /// </summary>
    /// <param name="writes">The requested writes.</param>
    /// <param name="cancellationToken">Cancellation for the batch.</param>
    /// <returns>One result per request, in request order.</returns>
    /// <remarks>
    ///     <para>
    ///     <b>Never throws, and one bad request never fails the batch.</b> Each slot carries its own
    ///     status, matching <see cref="ReadAsync"/>: failing the whole call would let a single stale
    ///     reference reject every other write in the same OPC UA request.
    ///     </para>
    ///     <para>
    ///     <b>Good means "published", not "applied".</b> MQTT has no application-level write
    ///     acknowledgement, so a successful PUBLISH is the strongest truth available at this seam —
    ///     the same optimistic-Good posture Galaxy's fire-and-forget write takes. Device-level truth
    ///     arrives afterwards, as the echoed value on the tag's read topic, and corrects the node
    ///     through the ordinary ingest path. A write whose PUBLISH itself fails is reported Bad: the
    ///     optimism covers what the device did with the command, never whether it was sent.
    ///     </para>
    ///     <para>
    ///     <b>Consequence, inherited knowingly (see #5, write-outcome self-correction):</b> because the
    ///     transport cannot report device-level rejection, this driver can never surface a failed
    ///     <em>device</em> write — exactly the property Galaxy has. A device that silently ignores a
    ///     command leaves the node reading whatever it last published, which is the honest picture; a
    ///     device that has no echo topic at all leaves the operator with no confirmation, which is a
    ///     property of the plant's topic design rather than of this code.
    ///     </para>
    /// </remarks>
    public async Task<IReadOnlyList<WriteResult>> WriteAsync(
        IReadOnlyList<WriteRequest> writes,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(writes);

        var results = new WriteResult[writes.Count];
        var connection = _connection;

        for (var i = 0; i < writes.Count; i++)
        {
            var w = writes[i];

            if (!TryResolve(w.FullReference, out var def))
            {
                results[i] = new WriteResult(StatusBadNodeIdUnknown);
                continue;
            }

            var error = MqttWriteEncoder.TryEncode(_options.Mode, def, w.Value, out var encoded);
            if (error != MqttWriteEncodeError.None)
            {
                results[i] = new WriteResult(MapEncodeError(error));
                continue;
            }

            if (connection is null)
            {
                // Not yet connected, or torn down mid-rebuild. Reported rather than queued: a write
                // buffered now and flushed at an unbounded later moment is the replay hazard
                // WriteIdempotent exists to prevent, decided by the driver instead of the operator.
                results[i] = new WriteResult(StatusBadCommunicationError);
                continue;
            }

            try
            {
                await connection
                    .PublishAsync(encoded.Topic, encoded.Payload, MqttWriteEncoder.WriteQos, retain: false, cancellationToken)
                    .ConfigureAwait(false);
                results[i] = new WriteResult(0u);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                // Caller-cancelled: fill the remainder rather than throwing, so the batch shape holds.
                for (var j = i; j < writes.Count; j++)
                {
                    results[j] = new WriteResult(StatusBadCommunicationError);
                }

                return results;
            }
            catch (Exception ex)
            {
                _logger?.LogWarning(
                    ex,
                    "MQTT driver '{DriverId}': write to '{Reference}' could not be published on '{Topic}'.",
                    _driverInstanceId,
                    w.FullReference,
                    encoded.Topic);
                results[i] = new WriteResult(StatusBadCommunicationError);
            }
        }

        return results;
    }

    /// <summary>Maps an encode refusal to the status an OPC UA client sees.</summary>
    /// <param name="error">The encode outcome.</param>
    /// <returns>The OPC UA status code.</returns>
    private static uint MapEncodeError(MqttWriteEncodeError error) => error switch
    {
        MqttWriteEncodeError.NoWriteDestination => StatusBadNotWritable,
        MqttWriteEncodeError.UnsupportedDataType => StatusBadNotWritable,
        _ => StatusBadTypeMismatch,
    };

    // ---- IHostConnectivityProbe ----

    /// <summary>The single broker this driver instance talks to, as <c>host:port</c>.</summary>
    public string HostName => $"{_options.Host}:{_options.Port}";

    /// <inheritdoc />
    public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
    {
        lock (_hostLock)
        {
            return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
        }
    }

    // ---- IRediscoverable ----

    /// <summary>
    ///     Raises <see cref="OnRediscoveryNeeded"/>. Nothing in <see cref="MqttMode.Plain"/> calls it
    ///     — the authored set only changes by redeploy. It exists for the Sparkplug B path, where a
    ///     DBIRTH can introduce metrics the previous birth certificate did not carry.
    /// </summary>
    /// <param name="reason">Driver-supplied reason for the diagnostic log.</param>
    /// <param name="scopeHint">Optional subtree hint; <see langword="null"/> means the whole tree.</param>
    internal void RaiseRediscoveryNeeded(string reason, string? scopeHint = null)
        => OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, scopeHint));

    /// <summary>
    ///     The rediscovery decision: a birth changed what <see cref="DiscoverAsync"/> can produce only
    ///     when it is for an <b>authored</b> scope <b>and</b> its metric-name set differs from the last
    ///     one seen for that scope. Anything else is dropped.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Both conditions are load-bearing, and both are anti-storm.</b> Rediscovery triggers an
    ///         address-space rebuild — expensive and disruptive on a running server. An edge node
    ///         re-births freely: on its own schedule, on every reconnect (the ingestor's late-join
    ///         rebirth fan-out), and once per rebirth NCMD its gap policy sends. Firing on each of those
    ///         is a permanent rebuild loop against a plant that is behaving perfectly. Firing on
    ///         <i>changed metric set only</i> is the primary defence; the authored-scope filter is the
    ///         second, and removes the whole class of churn sourced from devices this deployment does
    ///         not read (see <see cref="_authoredScopes"/>).
    ///     </para>
    ///     <para>
    ///         <b>The signature is order-insensitive</b> — ordered-distinct, ordinal. Nothing obliges an
    ///         edge node to keep its birth's metric ordering stable, and a reordered birth describes the
    ///         same address space; comparing the raw sequence would leak the storm back in.
    ///     </para>
    ///     <para>
    ///         <b>No additional debounce or coalescing.</b> One was considered and rejected: the change
    ///         gate already collapses the unbounded sources (rebirth timers, reconnect floods, NCMD
    ///         answers) to zero, and what survives it is a genuine metric-set change — of which a given
    ///         scope has a handful in its lifetime, each one a real edit to the served tree. A timer on
    ///         top would only delay a rebuild that must happen, and would need its own trailing-edge
    ///         flush to avoid dropping the last change entirely. If a consumer ever needs coalescing
    ///         across many scopes birthing at once (a whole-plant restart), that belongs in the consumer,
    ///         which alone knows what a rebuild costs it — today there is no such consumer at all.
    ///     </para>
    ///     <para>
    ///         <b>Runs on MQTTnet's dispatcher thread</b> (via the ingestor's own contained raise), so it
    ///         does no I/O and never throws — <see cref="SparkplugIngestor"/> catches a throwing
    ///         subscriber, but relying on that would still have cost the rest of the birth's fan-out.
    ///     </para>
    /// </remarks>
    private void OnBirthObserved(object? sender, SparkplugBirthObservedEventArgs e)
    {
        if (!_authoredScopes.Contains(e.Scope))
        {
            // A birth for a scope no authored tag binds cannot change the authored-only discovery tree.
            return;
        }

        var signature = BirthSignature(e.MetricNames);
        if (_birthSignatures.TryGetValue(e.Scope, out var previous)
            && string.Equals(previous, signature, StringComparison.Ordinal))
        {
            return; // Same metric set as last time — nothing to rediscover.
        }

        _birthSignatures[e.Scope] = signature;

        var reason = previous is null
            ? $"Sparkplug birth observed for '{e.Scope}' ({e.MetricNames.Count} metric(s))"
            : $"Sparkplug rebirth changed the metric set for '{e.Scope}' ({e.MetricNames.Count} metric(s))";

        _logger?.LogInformation(
            "MQTT driver '{DriverId}': {Reason}; requesting rediscovery.",
            _driverInstanceId,
            reason);

        // ScopeHint names the folder DiscoverAsync actually streams under — deliberately NOT the
        // Sparkplug scope path. The discovered tree is flat (one 'Mqtt' folder of RawPath-named
        // variables); it has no edge-node/device folders, so a hint of 'Plant1/EdgeA/Filler1' would
        // name a subtree that does not exist and a consumer scoping a rebuild on it would rebuild
        // nothing. The scope belongs in the Reason, which is the documented diagnostic field.
        RaiseRediscoveryNeeded(reason, DiscoveryFolderName);
    }

    /// <summary>An order-insensitive, ordinal signature of a birth's metric-name set.</summary>
    /// <param name="metricNames">The names the birth declared.</param>
    /// <returns>The comparable signature.</returns>
    private static string BirthSignature(IReadOnlyList<string> metricNames)
        => string.Join(
            '\u0001', // A separator no Sparkplug metric name can contain, so two sets cannot alias.
            metricNames.Distinct(StringComparer.Ordinal).Order(StringComparer.Ordinal));

    // ---- disposal ----

    /// <inheritdoc />
    public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();

    /// <summary>
    ///     Performs the same teardown as <see cref="ShutdownAsync"/>, <b>under the same lifecycle
    ///     gate</b>, so a caller that uses <c>await using</c> without an explicit shutdown does not
    ///     leak a live broker session.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>The gate wait is the whole point, not ceremony.</b> An ungated dispose that ran
    ///         while <see cref="ReinitializeAsync"/>'s rebuild branch held the gate would observe
    ///         <c>_connection == null</c> — the rebuild has torn the old session down but not yet
    ///         assigned the new one — do nothing, and set <c>_disposed</c>. The rebuild would then
    ///         complete, assign a <i>live, connected</i> connection and start a host-probe loop that
    ///         no later dispose can ever reach, because the disposed guard short-circuits them all.
    ///         That is the orphaned-connection class <see cref="MqttConnection"/>'s own remarks
    ///         describe having already fixed once. Waiting for the gate makes this dispose tear down
    ///         whatever the in-flight operation ends up assigning.
    ///     </para>
    ///     <para>
    ///         <c>_disposed</c> is set <b>before</b> the wait, so an in-flight
    ///         <see cref="InitializeCoreAsync"/> sees it at its post-connect re-check and refuses to
    ///         publish the connection at all — which is what closes the residual window on the
    ///         bounded-timeout fallback path below.
    ///     </para>
    ///     <para>
    ///         Separately (and for an unrelated reason): the gate object itself is deliberately never
    ///         <c>Dispose()</c>d — same call as <see cref="MqttConnection"/>'s semaphores. A caller
    ///         that disposes before its last <see cref="ShutdownAsync"/> would otherwise get an
    ///         <see cref="ObjectDisposedException"/> naming <see cref="SemaphoreSlim"/>, which says
    ///         nothing useful; the gate holds no timer and no surviving registration, so leaving it
    ///         undisposed costs nothing.
    ///     </para>
    /// </remarks>
    /// <returns>A task that represents the asynchronous dispose.</returns>
    public async ValueTask DisposeAsync()
    {
        if (Interlocked.Exchange(ref _disposed, 1) != 0)
        {
            return;
        }

        if (await _lifecycleGate.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds))
                .ConfigureAwait(false))
        {
            try
            {
                await TeardownAsync().ConfigureAwait(false);
            }
            finally
            {
                _lifecycleGate.Release();
            }

            return;
        }

        // Bounded even here: a wedged initialize must not turn dispose into a hang. The post-connect
        // disposed re-check in InitializeCoreAsync is what stops that operation from publishing a
        // connection behind this teardown's back.
        _logger?.LogWarning(
            "MQTT driver '{DriverId}': lifecycle gate still held at dispose; tearing the session down anyway.",
            _driverInstanceId);
        await TeardownAsync().ConfigureAwait(false);
    }

    // ---- internals ----

    /// <summary>The gate-held body of <see cref="InitializeAsync"/>.</summary>
    private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        WriteHealth(new DriverHealth(DriverState.Initializing, ReadHealth().LastSuccessfulRead, null));
        try
        {
            // Tear down whatever this driver already holds BEFORE building a new session. Both fields
            // published below (_connection, _primaryHost) are IAsyncDisposable and were previously
            // overwritten, so a second initialize on a live driver stranded a connected, supervised
            // broker session — plus a whole second primary-host connection — that no later dispose
            // could reach. Reachable without any teardown of ours: DriverInstanceActor re-runs
            // InitializeAsync with no ShutdownAsync when an ApplyDelta lands while the driver is in
            // Connecting/Reconnecting (AdoptConfigDuringInit). Idempotent, so the ReinitializeAsync
            // rebuild branch calling TeardownAsync first costs a no-op, and a first initialize sees
            // three nulls.
            await TeardownAsync().ConfigureAwait(false);

            // A blank / unusable blob keeps the constructor's options rather than degrading the
            // driver to defaults pointing at localhost.
            var parsed = ParseOptions(driverConfigJson);
            if (parsed is not null)
            {
                AdoptOptions(parsed);
            }

            RegisterAuthoredTags(_options);

            var connection = new MqttConnection(_options, _driverInstanceId, _logger);

            // MUST precede the connect: this wires message delivery AND the reconnect re-subscribe.
            // The handler is passed through unwrapped on purpose — its throw-on-total-failure is what
            // tears a deaf session down and retries it.
            if (_sparkplug is { } attaching)
            {
                attaching.AttachTo(connection);
            }
            else
            {
                _subscriptions.AttachTo(connection);
            }

            if (BeforeConnectHookForTests is { } hook)
            {
                await hook(cancellationToken).ConfigureAwait(false);
            }

            await connection.ConnectAsync(cancellationToken).ConfigureAwait(false);

            // Re-check disposal AFTER the connect, before publishing the connection. DisposeAsync
            // normally waits for this gate, but its wait is bounded — on the wedged-gate fallback
            // path it can run concurrently with this method, and assigning a live connection behind
            // a completed teardown is exactly the orphaned-session bug. Losing the race means
            // disposing what we just built, never leaking it.
            if (Volatile.Read(ref _disposed) != 0)
            {
                await connection.DisposeAsync().ConfigureAwait(false);
                throw new ObjectDisposedException(nameof(MqttDriver));
            }

            // Sparkplug subscribes ONCE, here, rather than on the OPC UA server's first subscribe: the
            // driver must see birth certificates whether or not anything is subscribed yet, because a
            // birth is where every alias and datatype comes from. This also issues the late-join
            // rebirth (§3.6 #5) — the initial connect does not fire Reconnected.
            //
            // It runs BEFORE `_connection` is published, and disposes the session it was given if it
            // fails. Publishing first would leave a live, supervised connection behind a throw: the
            // driver-host resilience layer re-runs this method, the retry overwrites `_connection`, and
            // the first session becomes an orphan no later dispose can reach — the exact leak
            // MqttConnection's own remarks describe having already fixed once.
            if (_sparkplug is { } sparkplug)
            {
                try
                {
                    await sparkplug.EstablishAsync(cancellationToken).ConfigureAwait(false);
                }
                catch
                {
                    await connection.DisposeAsync().ConfigureAwait(false);
                    throw;
                }
            }

            _connection = connection;

            WriteHealth(new DriverHealth(DriverState.Healthy, connection.LastMessageUtc, null));
            TransitionHost(HostState.Running);
            StartHostProbe();
            await StartPrimaryHostAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            WriteHealth(new DriverHealth(DriverState.Faulted, ReadHealth().LastSuccessfulRead, ex.Message));
            TransitionHost(HostState.Faulted);
            throw;
        }
    }

    /// <summary>
    ///     Deserializes a driver-config blob. Returns <see langword="null"/> — never throws — when the
    ///     blob is blank or unusable, so both the initialize fallback and the reinitialize
    ///     keep-running-config rule read off one answer.
    /// </summary>
    private MqttDriverOptions? ParseOptions(string driverConfigJson)
    {
        if (string.IsNullOrWhiteSpace(driverConfigJson))
        {
            return null;
        }

        try
        {
            // MqttJson.Options — the ONE shared instance (factory / probe / this / browser),
            // deliberately: enums must round-trip by NAME. A second, divergent
            // JsonSerializerOptions is this repo's documented systemic enum bug, where an
            // AdminUI-authored config with a string enum field faults the driver.
            var parsed = JsonSerializer.Deserialize<MqttDriverOptions>(driverConfigJson, MqttJson.Options);
            return parsed is null ? null : Normalize(parsed);
        }
        catch (JsonException ex)
        {
            _logger?.LogWarning(
                ex,
                "MQTT driver '{DriverId}': driver config JSON is invalid.",
                _driverInstanceId);
            return null;
        }
        catch (NotSupportedException ex)
        {
            _logger?.LogWarning(
                ex,
                "MQTT driver '{DriverId}': driver config JSON could not be bound to the options shape.",
                _driverInstanceId);
            return null;
        }
    }

    /// <summary>
    ///     Replaces out-of-range duration knobs with their documented defaults, once, at the single seam
    ///     every options instance this driver holds passes through (the constructor and
    ///     <see cref="ParseOptions"/>).
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <see cref="MqttDriverOptions"/> carries <c>[Range]</c> annotations, but nothing validates
    ///         them on the runtime path: the blob is deserialized straight out of the deploy artifact, so
    ///         a config authored (or hand-edited) with <c>connectTimeoutSeconds: 0</c> or a negative value
    ///         reaches the driver unchecked. Both are load-bearing, and neither fails where an operator
    ///         would look:
    ///     </para>
    ///     <list type="bullet">
    ///         <item>
    ///             zero makes every <c>new CancellationTokenSource(TimeSpan.Zero)</c> in the connect,
    ///             subscribe and publish paths fire immediately, so the driver can never connect and
    ///             reports a timeout against a perfectly healthy broker;
    ///         </item>
    ///         <item>
    ///             a negative value is outside <see cref="SemaphoreSlim.WaitAsync(TimeSpan)"/>'s and
    ///             <see cref="CancellationTokenSource"/>'s legal range, so it throws
    ///             <see cref="ArgumentOutOfRangeException"/> out of <see cref="ShutdownAsync"/> and
    ///             <see cref="DisposeAsync"/> — the two paths whose whole job is to not fail.
    ///         </item>
    ///     </list>
    ///     <para>
    ///         Normalising here rather than at each use site keeps one answer for the whole driver: the
    ///         value also rides into <see cref="MqttConnection"/>, <see cref="SparkplugPrimaryHost"/> and
    ///         <c>MqttClientOptionsFactory</c> through <see cref="_options"/>. <c>MaxPayloadBytes</c> and
    ///         the reconnect backoffs are deliberately left alone — the manager's constructor and
    ///         <see cref="MqttConnection.NextBackoff"/> already clamp those at their own use site.
    ///     </para>
    /// </remarks>
    private static MqttDriverOptions Normalize(MqttDriverOptions options)
    {
        var connectTimeout = options.ConnectTimeoutSeconds > 0
            ? options.ConnectTimeoutSeconds
            : DefaultConnectTimeoutSeconds;

        // Zero keep-alive is legal MQTT ("no keep-alive"); only a negative one is nonsense.
        var keepAlive = options.KeepAliveSeconds >= 0 ? options.KeepAliveSeconds : DefaultKeepAliveSeconds;

        return connectTimeout == options.ConnectTimeoutSeconds && keepAlive == options.KeepAliveSeconds
            ? options
            : options with { ConnectTimeoutSeconds = connectTimeout, KeepAliveSeconds = keepAlive };
    }

    /// <summary>
    ///     Adopts new options, rebuilding the subscription manager only when a setting it captures at
    ///     construction actually changed. The rebuilt manager inherits the driver-owned
    ///     <see cref="LastValueCache"/>, so adopting new options never blanks observed values.
    /// </summary>
    private void AdoptOptions(MqttDriverOptions next)
    {
        if (!SameIngest(next, _options))
        {
            _subscriptions = CreateSubscriptionManager(next);
            _sparkplug = CreateSparkplugIngestor(next);
        }

        _options = next;
    }

    /// <summary>Builds a manager over the driver-owned cache and bridges its notifications to <see cref="OnDataChange"/>.</summary>
    private MqttSubscriptionManager CreateSubscriptionManager(MqttDriverOptions options)
    {
        var manager = new MqttSubscriptionManager(
            options,
            _driverInstanceId,
            transport: null,
            logger: _logger,
            cache: _values,
            maxPayloadBytes: options.MaxPayloadBytes);

        // The published reference is the RawPath the manager already stamped on the args; the driver
        // only re-raises with itself as sender.
        manager.OnDataChange += (_, args) => OnDataChange?.Invoke(this, args);
        return manager;
    }

    /// <summary>
    ///     Builds the Sparkplug B ingest path for <paramref name="options"/>, or <see langword="null"/>
    ///     outside <see cref="MqttMode.SparkplugB"/>. Shares the driver-owned
    ///     <see cref="LastValueCache"/> for the same reason the plain manager does: a rebuild for
    ///     changed settings must not blank every observed value.
    /// </summary>
    private SparkplugIngestor? CreateSparkplugIngestor(MqttDriverOptions options)
    {
        if (options.Mode != MqttMode.SparkplugB)
        {
            return null;
        }

        // The flag no longer reads as if it does something it does not (Gitea #511) — but a request it
        // CANNOT honour still warns loudly rather than being dropped, because the operator's plant is
        // waiting on a STATE message that will never arrive.
        if (SparkplugPrimaryHost.DescribeUnusable(options) is { } unusable)
        {
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': {Reason} Edge nodes gated on this host id will not see it come online.",
                _driverInstanceId,
                unusable);
        }

        var ingestor = new SparkplugIngestor(
            options,
            _driverInstanceId,
            subscribeTransport: null,
            publishTransport: null,
            logger: _logger,
            cache: _values,
            maxPayloadBytes: options.MaxPayloadBytes);

        // The published reference is the RawPath the ingestor already stamped on the args; the driver
        // only re-raises with itself as sender.
        ingestor.OnDataChange += (_, args) => OnDataChange?.Invoke(this, args);

        // The ingestor detects a birth; the driver decides whether it changed the address space. Wired
        // here rather than in InitializeAsync so a driver rebuilt for a config delta re-arms it — an
        // ingest rebuild that silently dropped this subscription would leave a Sparkplug driver whose
        // datatypes fill in and whose rediscovery never fires again, with nothing to show for it.
        ingestor.BirthObserved += OnBirthObserved;
        return ingestor;
    }

    /// <summary>Registers the authored raw tags wholesale and refreshes the discovery enumeration set.</summary>
    private void RegisterAuthoredTags(MqttDriverOptions options)
    {
        // Wholesale into the LIVE path only. Feeding Sparkplug tags to the plain parser would log a
        // "did not map" warning for every one of them (they carry no topic) and vice versa.
        var mapped = _sparkplug is { } sparkplug
            ? sparkplug.Register(options.RawTags)
            : _subscriptions.Register(options.RawTags);

        // Enumerate in authoring order; the manager owns which of these actually mapped, and
        // DiscoverAsync skips the rest.
        _authoredRawPaths = [.. options.RawTags.Select(t => t.RawPath)];
        _authoredScopes = BuildAuthoredScopes();

        // Drop change-gate memory for scopes this deploy stopped authoring — the same housekeeping
        // SparkplugIngestor.Register does for its trackers, and for the same reason: without it a scope
        // nobody reads keeps a slot across every future redeploy. Re-authoring such a scope later means
        // its next birth reads as new and fires once, which is correct — a newly authored scope IS a
        // change to the discovered tree.
        foreach (var scope in _birthSignatures.Keys)
        {
            if (!_authoredScopes.Contains(scope))
            {
                _birthSignatures.TryRemove(scope, out _);
            }
        }

        // Gitea #541. Derived from the driver's OWN resolution path rather than from either manager's
        // internal bookkeeping, so what is reported skipped is exactly what read/subscribe will fail to
        // find — the two cannot drift, and one branch covers both ingest paths.
        _skippedTags.BeginRebuild(options.RawTags.Count);
        foreach (var entry in options.RawTags)
        {
            if (!TryResolve(entry.RawPath, out _))
                _skippedTags.Skip(entry.RawPath, "TagConfig did not map to a definition");
        }
        _skippedTags.Commit();

        if (mapped != options.RawTags.Count)
        {
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': {Mapped} of {Authored} authored tag(s) mapped to a definition; "
                + "the rest will report BadNodeIdUnknown.",
                _driverInstanceId,
                mapped,
                options.RawTags.Count);
        }
    }

    /// <summary>
    ///     The Sparkplug scopes the currently registered authored tags bind, derived from the same
    ///     definitions <see cref="DiscoverAsync"/> enumerates — empty outside
    ///     <see cref="MqttMode.SparkplugB"/>, and empty for any tag whose TagConfig did not map.
    /// </summary>
    /// <returns>The authored scope set.</returns>
    private FrozenSet<SparkplugScope> BuildAuthoredScopes()
    {
        if (_sparkplug is null)
        {
            return FrozenSet<SparkplugScope>.Empty;
        }

        var scopes = new HashSet<SparkplugScope>();
        foreach (var rawPath in _authoredRawPaths)
        {
            if (TryResolve(rawPath, out var def)
                && def.GroupId is { } groupId
                && def.EdgeNodeId is { } edgeNodeId)
            {
                scopes.Add(new SparkplugScope(groupId, edgeNodeId, def.DeviceId));
            }
        }

        return scopes.ToFrozenSet();
    }

    /// <summary>
    ///     Whether two option sets describe the same broker session. Note the explicit
    ///     <see cref="object.Equals(object?)"/>: the identity projections are <c>object</c>-typed
    ///     anonymous instances, so <c>!=</c> would compare <b>references</b> and report "changed"
    ///     every single time — which would tear down and rebuild a perfectly good session on every
    ///     redeploy, and (via <see cref="SameIngest"/>) swap in a subscription manager that is not
    ///     attached to the live connection.
    /// </summary>
    private static bool SameSession(MqttDriverOptions a, MqttDriverOptions b)
        => SessionIdentity(a).Equals(SessionIdentity(b));

    /// <summary>
    ///     Whether two option sets produce an identical ingest path — <see cref="MqttSubscriptionManager"/>
    ///     in Plain mode, <see cref="SparkplugIngestor"/> in Sparkplug B.
    /// </summary>
    /// <remarks>
    ///     ⚠️ <b>Every setting an ingest path captures at construction MUST be named by
    ///     <see cref="IngestIdentity"/>.</b> A setting that is captured but unnamed makes a delta
    ///     changing it compare "same ingest", get applied in place, and silently never reach the
    ///     component that needed rebuilding — the driver would keep subscribing to the OLD Sparkplug
    ///     group id and report perfect health. Task 21 closed the Sparkplug half of that gap by naming
    ///     the whole <see cref="MqttSparkplugOptions"/> record (it has value equality, so a member
    ///     added there is covered automatically — which is the point of naming the record rather than
    ///     enumerating its fields).
    /// </remarks>
    private static bool SameIngest(MqttDriverOptions a, MqttDriverOptions b)
        => IngestIdentity(a).Equals(IngestIdentity(b));

    /// <summary>
    ///     The settings whose change invalidates the live broker session. Projected onto an anonymous
    ///     type (structural <c>Equals</c>) rather than comparing the whole options record, whose
    ///     <c>RawTags</c> list would make every comparison unequal by reference.
    /// </summary>
    private static object SessionIdentity(MqttDriverOptions o) => new
    {
        o.Host,
        o.Port,
        o.ClientId,
        o.UseTls,
        o.AllowUntrustedServerCertificate,
        o.CaCertificatePath,
        o.Username,
        o.Password,
        o.ProtocolVersion,
        o.CleanSession,
        o.KeepAliveSeconds,
        o.ConnectTimeoutSeconds,
        o.ReconnectMinBackoffSeconds,
        o.ReconnectMaxBackoffSeconds,
        Ingest = IngestIdentity(o),
    };

    /// <summary>The settings the live ingest path captures at construction.</summary>
    /// <remarks>
    ///     <see cref="MqttSparkplugOptions"/> is named as a whole record rather than field-by-field: it
    ///     has structural value equality, so a member added to it in a later task is covered here with
    ///     no edit — and the failure mode of forgetting one is silent (see <see cref="SameIngest"/>).
    /// </remarks>
    private static object IngestIdentity(MqttDriverOptions o) => new
    {
        o.Mode,
        DefaultQos = o.Plain?.DefaultQos,
        o.MaxPayloadBytes,
        o.Sparkplug,
    };

    /// <summary>
    ///     Opens the Sparkplug primary-host claim when this configuration asks for one (Gitea #511).
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Runs last, and can never fail the initialize.</b> The claim is an advisory assertion
    ///         to other applications on the broker; the ingest path is this driver's data plane. A
    ///         broker that accepts the ingest session but rejects a publish on the STATE topic — an ACL
    ///         that does not grant <c>spBv1.0/STATE/#</c> is the ordinary way this happens — must cost
    ///         the operator a warning and a plant-visible missing host, not every tag's value.
    ///         <see cref="SparkplugPrimaryHost.StartAsync"/> already swallows and retries; the catch
    ///         here covers the rest so no future edit can turn this into a fault path by accident.
    ///     </para>
    /// </remarks>
    private async Task StartPrimaryHostAsync(CancellationToken cancellationToken)
    {
        if (!SparkplugPrimaryHost.CanClaim(_options))
        {
            return;
        }

        try
        {
            var host = new SparkplugPrimaryHost(_options, _driverInstanceId, _primaryRoleSource, _logger);
            _primaryHost = host;
            await host.StartAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            _logger?.LogWarning(
                ex,
                "MQTT driver '{DriverId}': the Sparkplug primary-host claim could not be started; ingest is unaffected.",
                _driverInstanceId);
        }
    }

    /// <summary>Shared teardown for <see cref="ShutdownAsync"/>, <see cref="DisposeAsync"/> and the session rebuild.</summary>
    private async Task TeardownAsync()
    {
        // Stand the host claim down FIRST, while the broker is still reachable and before the ingest
        // session goes. Its explicit retained OFFLINE is what makes the transition immediate for edge
        // nodes; leaving it to the Will costs them up to 1.5 keep-alive periods of believing a host
        // that has already stopped is still online.
        var primaryHost = Interlocked.Exchange(ref _primaryHost, null);
        if (primaryHost is not null)
        {
            await primaryHost.DisposeAsync().ConfigureAwait(false);
        }

        var probe = Interlocked.Exchange(ref _hostProbeCts, null);
        if (probe is not null)
        {
            await probe.CancelAsync().ConfigureAwait(false);
            probe.Dispose();
        }

        var connection = Interlocked.Exchange(ref _connection, null);
        if (connection is not null)
        {
            await connection.DisposeAsync().ConfigureAwait(false);
        }
    }

    /// <summary>
    ///     Starts the host-connectivity poll. <see cref="MqttConnection"/> exposes state but no
    ///     state-changed event, and an <see cref="IHostConnectivityProbe"/> whose event never fires is
    ///     a capability that looks wired and is inert — so the transition signal is sampled here.
    /// </summary>
    private void StartHostProbe()
    {
        var cts = new CancellationTokenSource();
        var previous = Interlocked.Exchange(ref _hostProbeCts, cts);
        if (previous is not null)
        {
            // Cancel before disposing: disposing a live source leaves its loop running against a
            // token that will never be cancelled, and a second probe would fight the first.
            previous.Cancel();
            previous.Dispose();
        }

        _ = Task.Run(() => HostProbeLoopAsync(cts.Token), cts.Token);
    }

    private async Task HostProbeLoopAsync(CancellationToken ct)
    {
        try
        {
            while (!ct.IsCancellationRequested)
            {
                await Task.Delay(HostProbeInterval, ct).ConfigureAwait(false);

                var connection = _connection;
                if (connection is null)
                {
                    continue;
                }

                TransitionHost(connection.State switch
                {
                    MqttConnectionState.Connected => HostState.Running,
                    MqttConnectionState.Reconnecting or MqttConnectionState.Disconnected => HostState.Stopped,
                    MqttConnectionState.Faulted => HostState.Faulted,
                    _ => HostState.Unknown,
                });
            }
        }
        catch (OperationCanceledException)
        {
            // Shutdown.
        }
        catch (Exception ex)
        {
            _logger?.LogWarning(ex, "MQTT driver '{DriverId}': host connectivity probe stopped.", _driverInstanceId);
        }
    }

    /// <summary>Publishes a host state transition once; repeats of the current state are ignored.</summary>
    private void TransitionHost(HostState next)
    {
        HostState previous;
        lock (_hostLock)
        {
            if (_hostState == next)
            {
                return;
            }

            previous = _hostState;
            _hostState = next;
            _hostStateChangedUtc = DateTime.UtcNow;
        }

        OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, previous, next));
    }

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