lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs

using System.Text;
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;

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

/// <summary>
///     Shell-level tests for <see cref="MqttDriver"/>: authored-only discovery, the lifecycle
///     surface (<c>Reinitialize</c> / <c>Shutdown</c> / health / footprint / cache flush), and the
///     <c>IReadable</c> batch contract. Nothing here dials a broker — every test exercises the
///     connection-free half of the driver, which is exactly the half a chatty broker must not be
///     able to influence.
/// </summary>
[Trait("Category", "Unit")]
public sealed class MqttDriverDiscoveryTests
{
    /// <summary>
    ///     An authored TagConfig blob. <c>dataType</c> uses the real <see cref="DriverDataType"/>
    ///     member names — there is no <c>Double</c>; the 64-bit float is <c>Float64</c>.
    /// </summary>
    private static string TagJson(string topic, string dataType = "String", string payloadFormat = "Raw")
        => $$"""{"topic":"{{topic}}","payloadFormat":"{{payloadFormat}}","dataType":"{{dataType}}"}""";

    private static RawTagEntry Tag(string rawPath, string topic, string dataType = "String")
        => new(rawPath, TagJson(topic, dataType), WriteIdempotent: false);

    private static MqttDriver PlainDriver(params RawTagEntry[] tags)
        => new(new MqttDriverOptions { Mode = MqttMode.Plain, RawTags = tags }, "d", null);

    /// <summary>
    ///     Options aimed at a definitely-closed port so a connect is <b>refused immediately</b> rather
    ///     than hanging — the tests that must prove a rebuild actually dials need the dial to fail
    ///     fast, not to burn a connect deadline.
    /// </summary>
    private static MqttDriverOptions ClosedPortOptions(params RawTagEntry[] tags) => new()
    {
        Mode = MqttMode.Plain,
        Host = "127.0.0.1",
        Port = 1,
        UseTls = false,
        ConnectTimeoutSeconds = 2,
        RawTags = tags,
    };

    private static MqttDriver ClosedPortDriver(params RawTagEntry[] tags)
        => new(ClosedPortOptions(tags), "d", null);

    /// <summary>
    ///     Serializes a full options record as the reinitialize blob. Round-tripping the whole record
    ///     (rather than hand-writing a partial JSON object) guarantees that only the field the caller
    ///     changed differs — a hand-written delta silently omitting a session field would take the
    ///     rebuild branch for the wrong reason and the test would pass vacuously.
    /// </summary>
    private static string DeltaJson(MqttDriverOptions options)
        => System.Text.Json.JsonSerializer.Serialize(options, MqttJson.Options);

    /// <summary>
    ///     Plain mode replays ONLY the authored tag set: one variable per authored raw tag, a
    ///     single discovery pass (<see cref="DiscoveryRediscoverPolicy.Once"/>), and no online
    ///     discovery — the universal browser must never treat a broker's topic traffic as a
    ///     browsable address space.
    /// </summary>
    [Fact]
    public async Task DiscoverAsync_Plain_StreamsAuthoredTagsOnly_PolicyOnce()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
        var b = new CapturingAddressSpaceBuilder();

        await driver.DiscoverAsync(b, CancellationToken.None);

        b.Variables.Count.ShouldBe(1);
        b.Variables[0].Info.FullName.ShouldBe("Plant/Mqtt/dev1/Temp");
        driver.RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.Once);
        ((ITagDiscovery)driver).SupportsOnlineDiscovery.ShouldBeFalse();
    }

    /// <summary>
    ///     The falsifiability pin for authored-only discovery: a message that arrives on a topic no
    ///     authored tag names must not add anything to the discovered set. A driver that
    ///     auto-provisioned from broker traffic would grow the address space on every deploy against
    ///     a chatty broker.
    /// </summary>
    [Fact]
    public async Task DiscoverAsync_UnauthoredBrokerTraffic_DoesNotAppear()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        // Traffic this driver never authored — a neighbouring publisher on the same broker.
        driver.Subscriptions.HandleMessage("some/other/topic", Encoding.UTF8.GetBytes("42"), retained: false);
        driver.Subscriptions.HandleMessage("f/t/deeper", Encoding.UTF8.GetBytes("42"), retained: false);

        var b = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(b, CancellationToken.None);

        b.Variables.Select(v => v.Info.FullName).ShouldBe(["Plant/Mqtt/dev1/Temp"]);
    }

    /// <summary>A raw tag whose TagConfig does not map is skipped, never thrown, and never discovered.</summary>
    [Fact]
    public async Task DiscoverAsync_SkipsUnmappableTagConfig()
    {
        var driver = PlainDriver(
            Tag("Plant/Mqtt/dev1/Good", "f/t"),
            new RawTagEntry("Plant/Mqtt/dev1/Bad", "not-json", WriteIdempotent: false));

        var b = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(b, CancellationToken.None);

        b.Variables.Select(v => v.Info.FullName).ShouldBe(["Plant/Mqtt/dev1/Good"]);
    }

    /// <summary>Discovered MQTT variables are read-only — this driver has no <c>IWritable</c> leg.</summary>
    [Fact]
    public async Task DiscoverAsync_MarksVariablesViewOnly()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t", "Float64"));
        var b = new CapturingAddressSpaceBuilder();

        await driver.DiscoverAsync(b, CancellationToken.None);

        b.Variables[0].Info.SecurityClass.ShouldBe(SecurityClassification.ViewOnly);
        b.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.Float64);
    }

    /// <summary>
    ///     The falsifiability pin for the batch-read contract: a reference that is not an authored
    ///     tag degrades to <c>BadWaitingForInitialData</c> in its own slot rather than throwing and
    ///     failing every other reference in the same batch.
    /// </summary>
    [Fact]
    public async Task ReadAsync_UnknownReference_DoesNotThrow_AndDegradesPerRef()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
        driver.Subscriptions.HandleMessage("f/t", Encoding.UTF8.GetBytes("hot"), retained: false);

        var results = await driver.ReadAsync(
            ["Plant/Mqtt/dev1/Temp", "Plant/Mqtt/dev1/NeverAuthored"],
            CancellationToken.None);

        results.Count.ShouldBe(2);
        results[0].StatusCode.ShouldBe(0u);
        results[0].Value.ShouldBe("hot");
        results[1].StatusCode.ShouldBe(0x80320000u); // BadWaitingForInitialData
    }

    /// <summary>An empty batch is legal and returns an empty result, not an exception.</summary>
    [Fact]
    public async Task ReadAsync_EmptyBatch_ReturnsEmpty()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        var results = await driver.ReadAsync([], CancellationToken.None);

        results.ShouldBeEmpty();
    }

    /// <summary>
    ///     The falsifiability pin for the cache-flush contract: the last-value cache backs
    ///     <c>IReadable</c>, so flushing it would silently turn every subscribed node Bad under
    ///     memory pressure. <c>FlushOptionalCachesAsync</c> must leave it untouched.
    /// </summary>
    [Fact]
    public async Task FlushOptionalCachesAsync_LeavesLastValueCacheIntact()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
        driver.Subscriptions.HandleMessage("f/t", Encoding.UTF8.GetBytes("hot"), retained: false);

        await driver.FlushOptionalCachesAsync(CancellationToken.None);

        var results = await driver.ReadAsync(["Plant/Mqtt/dev1/Temp"], CancellationToken.None);
        results[0].StatusCode.ShouldBe(0u);
        results[0].Value.ShouldBe("hot");
    }

    /// <summary>A malformed reinitialize delta must never fault the driver, and must not lose the running config.</summary>
    [Fact]
    public async Task ReinitializeAsync_MalformedDelta_KeepsRunningConfig_AndDoesNotFault()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        await Should.NotThrowAsync(() => driver.ReinitializeAsync("{ this is not json", CancellationToken.None));

        driver.GetHealth().State.ShouldNotBe(DriverState.Faulted);

        var b = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(b, CancellationToken.None);
        b.Variables.Select(v => v.Info.FullName).ShouldBe(["Plant/Mqtt/dev1/Temp"]);
    }

    /// <summary>
    ///     A delta that changes only the authored tag set is applied in place — no teardown, no
    ///     reconnect — and is applied WHOLESALE: a tag the redeploy dropped stops being discovered.
    /// </summary>
    [Fact]
    public async Task ReinitializeAsync_TagOnlyDelta_AppliedInPlace_Wholesale()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        var pressure = System.Text.Json.JsonSerializer.Serialize(TagJson("f/p"));
        var flow = System.Text.Json.JsonSerializer.Serialize(TagJson("f/q"));
        var delta = $$"""
        {
          "mode": "Plain",
          "rawTags": [
            { "rawPath": "Plant/Mqtt/dev1/Pressure", "tagConfig": {{pressure}}, "writeIdempotent": false },
            { "rawPath": "Plant/Mqtt/dev1/Flow", "tagConfig": {{flow}}, "writeIdempotent": false }
          ]
        }
        """;

        await driver.ReinitializeAsync(delta, CancellationToken.None);

        var b = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(b, CancellationToken.None);

        b.Variables.Select(v => v.Info.FullName)
            .OrderBy(x => x, StringComparer.Ordinal)
            .ShouldBe(["Plant/Mqtt/dev1/Flow", "Plant/Mqtt/dev1/Pressure"]);
    }

    /// <summary>
    ///     A delta that changes the broker endpoint takes the <b>rebuild</b> branch: it must actually
    ///     dial the new endpoint (proven here by the refused connect), and a rebuild whose connect
    ///     fails must land <see cref="DriverState.Faulted"/> rather than letting the failure escape
    ///     with the health surface still claiming Healthy.
    /// </summary>
    [Fact]
    public async Task ReinitializeAsync_SessionChangingDelta_Rebuilds_AndFaultsOnUnreachableBroker()
    {
        var driver = ClosedPortDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        // Only Host differs from the ctor options — everything else round-trips identically.
        var delta = DeltaJson(ClosedPortOptions() with { Host = "127.0.0.2" });

        await Should.ThrowAsync<Exception>(() => driver.ReinitializeAsync(delta, CancellationToken.None));

        driver.GetHealth().State.ShouldBe(DriverState.Faulted);
    }

    /// <summary>
    ///     Pins that <c>IngestIdentity</c> is nested inside <c>SessionIdentity</c>: a delta touching
    ///     ONLY an ingest setting still rebuilds. Without the nesting it would be judged "same
    ///     session", applied in place, and the subscription manager that actually reads
    ///     <c>MaxPayloadBytes</c> would never be rebuilt — a config change that silently does nothing.
    /// </summary>
    [Fact]
    public async Task ReinitializeAsync_IngestOnlyDelta_AlsoRebuilds()
    {
        var driver = ClosedPortDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        var delta = DeltaJson(ClosedPortOptions() with { MaxPayloadBytes = 4096 });

        // Reaching the (refused) connect at all is the proof the rebuild branch was taken; the
        // tag-only test above is the control that shows an in-place delta never dials.
        await Should.ThrowAsync<Exception>(() => driver.ReinitializeAsync(delta, CancellationToken.None));
    }

    /// <summary>
    ///     The falsifiability pin for the orphaned-connection class: <see cref="MqttDriver.DisposeAsync"/>
    ///     must serialize behind an in-flight lifecycle operation, not race past it.
    /// </summary>
    /// <remarks>
    ///     An ungated dispose running while a rebuild holds the gate observes <c>_connection == null</c>
    ///     (the rebuild has torn the old session down but not yet assigned the new one), does nothing,
    ///     and sets the disposed flag — after which the rebuild assigns a live connection plus a
    ///     host-probe loop that no later dispose can reach. Here the driver is parked inside
    ///     <c>InitializeCoreAsync</c> at the pre-connect hook; the assertion is that dispose does not
    ///     complete while it is parked, and does complete once it is released.
    /// </remarks>
    [Fact]
    public async Task DisposeAsync_SerializesBehindAnInFlightLifecycleOperation()
    {
        var driver = ClosedPortDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        driver.BeforeConnectHookForTests = async _ =>
        {
            entered.TrySetResult();
            await release.Task;
        };

        var initialize = Task.Run(() => driver.InitializeAsync("", CancellationToken.None));
        await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));

        var dispose = driver.DisposeAsync().AsTask();

        // The gate is held by the parked initialize. Dispose must still be waiting. (The gate wait is
        // bounded by ConnectTimeoutSeconds = 2 s, so this 300 ms window is comfortably inside it —
        // an ungated dispose returns essentially instantly.)
        var raced = await Task.WhenAny(dispose, Task.Delay(TimeSpan.FromMilliseconds(300)));
        raced.ShouldNotBe(dispose, "DisposeAsync returned while a lifecycle operation held the gate");

        release.SetResult();
        await Should.ThrowAsync<Exception>(() => initialize); // connect refused on the closed port

        await dispose.WaitAsync(TimeSpan.FromSeconds(10));
        dispose.IsCompletedSuccessfully.ShouldBeTrue();
    }

    /// <summary>Identity is fixed at construction and must match the persisted <c>DriverInstance.DriverType</c>.</summary>
    [Fact]
    public void Identity_IsMqtt()
    {
        var driver = PlainDriver();

        driver.DriverType.ShouldBe("Mqtt");
        driver.DriverInstanceId.ShouldBe("d");
    }

    /// <summary>A driver that has never been initialized reports Unknown, not Healthy.</summary>
    [Fact]
    public void GetHealth_BeforeInitialize_IsUnknown()
    {
        PlainDriver().GetHealth().State.ShouldBe(DriverState.Unknown);
    }

    /// <summary>Plain mode never signals rediscovery — the authored set only changes by redeploy.</summary>
    [Fact]
    public async Task Plain_NeverRaisesRediscoveryNeeded()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));
        var fired = 0;
        ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);

        driver.Subscriptions.HandleMessage("f/t", Encoding.UTF8.GetBytes("hot"), retained: false);
        await driver.DiscoverAsync(new CapturingAddressSpaceBuilder(), CancellationToken.None);

        fired.ShouldBe(0);
    }

    /// <summary>The footprint tracks the authored table; an empty driver reports no driver-attributable bytes.</summary>
    [Fact]
    public void GetMemoryFootprint_TracksAuthoredTable()
    {
        PlainDriver().GetMemoryFootprint().ShouldBe(0);
        PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t")).GetMemoryFootprint().ShouldBeGreaterThan(0);
    }

    /// <summary>Shutdown on a never-initialized driver is a no-op, not a null-reference.</summary>
    [Fact]
    public async Task ShutdownAsync_BeforeInitialize_DoesNotThrow()
    {
        var driver = PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t"));

        await Should.NotThrowAsync(() => driver.ShutdownAsync(CancellationToken.None));
        driver.GetHealth().State.ShouldBe(DriverState.Unknown);
    }

    /// <summary>The single-broker connectivity probe names the configured endpoint.</summary>
    [Fact]
    public void GetHostStatuses_ReportsTheConfiguredBroker()
    {
        var driver = new MqttDriver(
            new MqttDriverOptions { Mode = MqttMode.Plain, Host = "broker.example", Port = 1883 },
            "d",
            null);

        var statuses = ((IHostConnectivityProbe)driver).GetHostStatuses();

        statuses.Count.ShouldBe(1);
        statuses[0].HostName.ShouldBe("broker.example:1883");
        statuses[0].State.ShouldBe(HostState.Unknown);
    }

    // =================================================================================
    // Task 22 — Sparkplug discovery policy, birth-filled datatypes, rediscovery trigger
    // =================================================================================

    /// <summary>
    ///     Sparkplug's discovered shape is not fully known at connect: a tag authored without a
    ///     <c>dataType</c> takes its type from the birth certificate, which arrives asynchronously.
    ///     That is precisely the <see cref="DiscoveryRediscoverPolicy.UntilStable"/> contract — the
    ///     host re-runs discovery until the captured set settles.
    /// </summary>
    [Fact]
    public void SparkplugMode_RediscoverPolicy_IsUntilStable()
        => SparkplugDriver().RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.UntilStable);

    /// <summary>
    ///     The mode gate, asserted from the other side: plain MQTT's authored set is fully known
    ///     synchronously, so it must stay <see cref="DiscoveryRediscoverPolicy.Once"/>. A blanket flip
    ///     to <c>UntilStable</c> would make every plain deployment re-run discovery on a timer for a
    ///     tree that can never change.
    /// </summary>
    [Fact]
    public void PlainMode_RediscoverPolicy_StaysOnce()
        => PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t")).RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.Once);

    /// <summary>
    ///     A DBIRTH for an authored device the driver has never held a birth for introduces the
    ///     metrics' datatypes, so the discovered tree genuinely changed: rediscovery fires.
    /// </summary>
    [Fact]
    public void NewDbirth_FiresOnRediscoveryNeeded()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
        var fired = 0;
        ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);

        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        fired.ShouldBeGreaterThan(0);
    }

    /// <summary>
    ///     <b>The primary anti-storm pin.</b> An edge node that re-births on a timer (or a flapping
    ///     connection driving the late-join rebirth) republishes the same metric set over and over. If
    ///     every birth fired rediscovery, each one would trigger an address-space rebuild — expensive
    ///     and disruptive on a running server, forever, for a tree that never changed.
    /// </summary>
    [Fact]
    public void IdenticalRebirth_DoesNotFireRediscovery()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        // Subscribe only AFTER the first birth so the count below is purely the rebirths'.
        var fired = 0;
        ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);

        for (var i = 0; i < 5; i++)
        {
            FeedNodeBirth(driver, seq: 0, bdSeq: (ulong)(i + 2));
            FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
        }

        fired.ShouldBe(0);
    }

    /// <summary>
    ///     The other half of the gate: a rebirth that actually changes the metric set (a metric added,
    ///     removed or renamed) DOES change what discovery can resolve, and must fire.
    /// </summary>
    [Fact]
    public void RebirthWithChangedMetricSet_FiresRediscovery()
    {
        var driver = SparkplugDriver(
            SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")),
            SpTag(SpPressPath, SpBlob(SpDevice, "Pressure")));
        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        var fired = 0;
        ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);

        FeedDeviceBirth(driver, seq: 2, ("Temperature", 5UL, TahuDataType.Float), ("Pressure", 6UL, TahuDataType.Float));

        fired.ShouldBe(1);
    }

    /// <summary>
    ///     Metric ORDER is not the metric SET. A birth that lists the same metrics in a different order
    ///     describes the same address space; firing on it would reintroduce the rebuild storm through
    ///     the back door, because nothing obliges an edge node to keep its birth order stable.
    /// </summary>
    [Fact]
    public void RebirthReorderingTheSameMetrics_DoesNotFireRediscovery()
    {
        var driver = SparkplugDriver(
            SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")),
            SpTag(SpPressPath, SpBlob(SpDevice, "Pressure")));
        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float), ("Pressure", 6UL, TahuDataType.Float));

        var fired = 0;
        ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);

        FeedDeviceBirth(driver, seq: 2, ("Pressure", 6UL, TahuDataType.Float), ("Temperature", 5UL, TahuDataType.Float));

        fired.ShouldBe(0);
    }

    /// <summary>
    ///     A birth for a device NO authored tag names cannot change <see cref="MqttDriver.DiscoverAsync"/>'s
    ///     output — discovery replays the authored set. A large plant DBIRTHs devices this deployment
    ///     never reads on their own schedules; firing for those is a rebuild storm sourced from traffic
    ///     the configuration has no opinion about.
    /// </summary>
    [Fact]
    public void BirthForAnUnauthoredDevice_DoesNotFireRediscovery()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
        FeedNodeBirth(driver);

        var fired = 0;
        ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);

        // Same (authored) edge node, a device nothing binds — the ingestor still applies the birth.
        FeedDeviceBirth(driver, "Filler99", seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        fired.ShouldBe(0);
    }

    /// <summary>
    ///     The rediscovery event names the subtree this driver's discovery actually emits (the
    ///     <c>Mqtt</c> folder) so a consumer scoping a rebuild on it scopes to exactly the tree that
    ///     changed, and carries the Sparkplug scope in the diagnostic reason.
    /// </summary>
    [Fact]
    public void RediscoveryEvent_ScopeHintIsTheDiscoveryFolder_ReasonNamesTheScope()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
        RediscoveryEventArgs? args = null;
        ((IRediscoverable)driver).OnRediscoveryNeeded += (_, e) => args = e;

        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        args.ShouldNotBeNull();
        args.ScopeHint.ShouldBe("Mqtt");
        args.Reason.ShouldContain($"{SpGroup}/{SpNode}/{SpDevice}");
    }

    /// <summary>
    ///     <b>The datatype fill-in.</b> <c>dataType</c> is optional in the Sparkplug tag shape — the
    ///     birth certificate declares it. Before any birth the tag can only report the record default;
    ///     once the DBIRTH lands, discovery must re-stream it with the type the birth declared.
    /// </summary>
    [Fact]
    public async Task DiscoverAsync_Sparkplug_UnauthoredDataType_FillsInFromTheBirth()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));

        var before = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(before, TestContext.Current.CancellationToken);

        // The node exists from the first pass — an authored tag is part of the declared configuration,
        // not something the plant grants by publishing.
        before.Variables.Select(v => v.Info.FullName).ShouldBe([SpTempPath]);
        before.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.String); // the record default

        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        var after = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(after, TestContext.Current.CancellationToken);

        after.Variables.Select(v => v.Info.FullName).ShouldBe([SpTempPath]);
        after.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.Float32);
    }

    /// <summary>
    ///     The authored <c>dataType</c> wins over the birth's — the same precedence the ingest path
    ///     applies when it coerces a value (<c>DataTypeAuthored ? authored : birth</c>). A discovery
    ///     surface that disagreed with the publish surface would report a type no value ever arrives as.
    /// </summary>
    [Fact]
    public async Task DiscoverAsync_Sparkplug_AuthoredDataType_WinsOverTheBirth()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature", "Int32")));
        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        var b = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(b, TestContext.Current.CancellationToken);

        b.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.Int32);
    }

    /// <summary>
    ///     A birth declaring a Sparkplug type this driver cannot map (DataSet / Template / PropertySet
    ///     / Unknown) must not blank the tag's type — <c>ToDriverDataType()</c> returns null for those,
    ///     and discovery falls back to the authored/default type rather than to nothing.
    /// </summary>
    [Fact]
    public async Task DiscoverAsync_Sparkplug_UnsupportedBirthType_FallsBackToTheDefault()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.DataSet));

        var b = new CapturingAddressSpaceBuilder();
        await driver.DiscoverAsync(b, TestContext.Current.CancellationToken);

        b.Variables.Count.ShouldBe(1);
        b.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.String);
    }

    /// <summary>
    ///     <b>The composition-root line.</b> Task 21 wires the ingestor's <c>OnDataChange</c> to the
    ///     driver's own in <c>CreateSparkplugIngestor</c>; every other Sparkplug test in the repo drives
    ///     <see cref="SparkplugIngestor"/> directly and would pass with that lambda deleted. This repo
    ///     has twice shipped a component correct in isolation and inert in production for exactly that
    ///     reason (<c>DeferredAddressSpaceSink</c>, <c>GatewayTagProvisioner</c>) — so the re-raise is
    ///     asserted through <see cref="MqttDriver"/>'s own event, under the RawPath.
    /// </summary>
    [Fact]
    public async Task SparkplugMode_DriverRepublishesIngestDataChanges_UnderTheRawPath()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature", "Float32")));
        await driver.SubscribeAsync([SpTempPath], TimeSpan.Zero, TestContext.Current.CancellationToken);

        object? sender = null;
        string? firedRef = null;
        object? firedValue = null;
        ((ISubscribable)driver).OnDataChange += (s, e) =>
        {
            sender = s;
            firedRef = e.FullReference;
            firedValue = e.Snapshot.Value;
        };

        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
        FeedDeviceData(driver, seq: 2, (5UL, 21.5f));

        firedRef.ShouldBe(SpTempPath);
        firedValue.ShouldBe(21.5f);

        // The driver re-raises with ITSELF as sender, not the ingestor — DriverHostActor attributes
        // publishes to the driver instance.
        sender.ShouldBeSameAs(driver);
    }

    /// <summary>
    ///     <c>ReadAsync</c> serves from the driver-owned <c>LastValueCache</c>, which the Sparkplug
    ///     ingestor is handed at construction. If the two ever stopped sharing one instance, every
    ///     Sparkplug read would answer <c>BadWaitingForInitialData</c> forever while subscriptions
    ///     looked perfectly healthy.
    /// </summary>
    [Fact]
    public async Task SparkplugMode_ReadAsyncServesTheIngestedValue()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature", "Float32")));

        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
        FeedDeviceData(driver, seq: 2, (5UL, 33.5f));

        var results = await driver.ReadAsync([SpTempPath], TestContext.Current.CancellationToken);

        results[0].Value.ShouldBe(33.5f);
        results[0].StatusCode.ShouldBe(0x00000000u);
    }

    /// <summary>
    ///     Subscribe/unsubscribe dispatch by mode. After an unsubscribe the reference is no longer
    ///     attributed to a handle, so ingest stops raising for it — while the cache keeps answering, so
    ///     a read still works. Both halves matter: a driver that kept publishing after unsubscribe
    ///     would leak notifications into a torn-down monitored item.
    /// </summary>
    [Fact]
    public async Task SparkplugMode_UnsubscribeStopsNotifications_ButNotReads()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature", "Float32")));
        var handle = await driver.SubscribeAsync([SpTempPath], TimeSpan.Zero, TestContext.Current.CancellationToken);

        FeedNodeBirth(driver);
        FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));

        var fired = 0;
        ((ISubscribable)driver).OnDataChange += (_, _) => Interlocked.Increment(ref fired);

        await driver.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
        FeedDeviceData(driver, seq: 2, (5UL, 7.5f));

        fired.ShouldBe(0);
        (await driver.ReadAsync([SpTempPath], TestContext.Current.CancellationToken))[0].Value.ShouldBe(7.5f);
    }

    /// <summary>
    ///     The mode gate from the other side — the reviewer's reading of the branching, turned into a
    ///     test. A Plain-mode driver builds NO Sparkplug ingestor at all, so no Sparkplug traffic can
    ///     reach it and a Sparkplug-shaped blob (which carries no <c>topic</c>) resolves to nothing.
    /// </summary>
    [Fact]
    public void PlainMode_BuildsNoSparkplugIngestor()
    {
        var driver = new MqttDriver(
            new MqttDriverOptions
            {
                Mode = MqttMode.Plain,
                RawTags = [SpTag(SpTempPath, SpBlob(SpDevice, "Temperature", "Float32"))],
            },
            "d",
            null);

        driver.Sparkplug.ShouldBeNull();
        driver.Subscriptions.TryResolve(SpTempPath, out _).ShouldBeFalse();
    }

    /// <summary>
    ///     …and the converse: a Sparkplug-mode driver registers into the Sparkplug path, not the plain
    ///     manager. Registering into both would double-warn on every deploy and, worse, let a plain
    ///     topic route a Sparkplug tag.
    /// </summary>
    [Fact]
    public void SparkplugMode_RegistersIntoTheSparkplugPathOnly()
    {
        var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature", "Float32")));

        driver.Sparkplug!.TryResolve(SpTempPath, out var def).ShouldBeTrue();
        def.MetricName.ShouldBe("Temperature");
        driver.Subscriptions.TryResolve(SpTempPath, out _).ShouldBeFalse();
    }

    // ---- Sparkplug fixtures ----

    private const string SpGroup = "Plant1";
    private const string SpNode = "EdgeA";
    private const string SpDevice = "Filler1";
    private const string SpTempPath = "Plant/Mqtt/spb/Filler1Temp";
    private const string SpPressPath = "Plant/Mqtt/spb/Filler1Press";

    /// <summary>A Sparkplug tag blob. <c>dataType</c> is omitted unless supplied — it is optional.</summary>
    private static string SpBlob(string? device, string metric, string? dataType = null)
    {
        var deviceKey = device is null ? "" : $"""{'"'}deviceId{'"'}:"{device}",""";
        var typeKey = dataType is null ? "" : $""","dataType":"{dataType}" """;
        return $$"""{"groupId":"{{SpGroup}}","edgeNodeId":"{{SpNode}}",{{deviceKey}}"metricName":"{{metric}}"{{typeKey}}}""";
    }

    private static RawTagEntry SpTag(string rawPath, string blob) => new(rawPath, blob, WriteIdempotent: false);

    private static MqttDriver SparkplugDriver(params RawTagEntry[] tags)
        => new(
            new MqttDriverOptions
            {
                Mode = MqttMode.SparkplugB,
                Sparkplug = new MqttSparkplugOptions { GroupId = SpGroup },
                RawTags = tags,
            },
            "d",
            null);

    /// <summary>
    ///     Feeds a real NBIRTH through the real codec + ingest state machine — no broker, and no
    ///     test-only shortcut on the driver. <c>Dispatch</c> is the method MQTTnet's dispatcher thread
    ///     reaches, so this exercises the production path end to end.
    /// </summary>
    private static void FeedNodeBirth(MqttDriver driver, ulong seq = 0, ulong bdSeq = 1)
    {
        var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
        payload.Metrics.Add(new Payload.Types.Metric
        {
            Name = "bdSeq",
            Datatype = (uint)TahuDataType.Uint64,
            LongValue = bdSeq,
        });
        payload.Metrics.Add(SpBirthMetric("Uptime", 7UL, TahuDataType.Int64));

        driver.Sparkplug!.Dispatch(
            new SparkplugTopic(SparkplugMessageType.NBIRTH, SpGroup, SpNode, null, null),
            SparkplugCodec.Decode(payload.ToByteArray()));
    }

    private static void FeedDeviceBirth(
        MqttDriver driver,
        ulong seq,
        params (string Name, ulong Alias, TahuDataType Type)[] metrics)
        => FeedDeviceBirth(driver, SpDevice, seq, metrics);

    private static void FeedDeviceBirth(
        MqttDriver driver,
        string device,
        ulong seq,
        params (string Name, ulong Alias, TahuDataType Type)[] metrics)
    {
        var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
        foreach (var (name, alias, type) in metrics)
        {
            payload.Metrics.Add(SpBirthMetric(name, alias, type));
        }

        driver.Sparkplug!.Dispatch(
            new SparkplugTopic(SparkplugMessageType.DBIRTH, SpGroup, SpNode, device, null),
            SparkplugCodec.Decode(payload.ToByteArray()));
    }

    /// <summary>
    ///     Feeds a real DDATA — alias-only metrics, exactly as a Sparkplug node publishes after a birth.
    /// </summary>
    private static void FeedDeviceData(MqttDriver driver, ulong seq, params (ulong Alias, float Value)[] metrics)
    {
        var payload = new Payload { Seq = seq, Timestamp = 1721822405000UL };
        foreach (var (alias, value) in metrics)
        {
            payload.Metrics.Add(new Payload.Types.Metric { Alias = alias, FloatValue = value });
        }

        driver.Sparkplug!.Dispatch(
            new SparkplugTopic(SparkplugMessageType.DDATA, SpGroup, SpNode, SpDevice, null),
            SparkplugCodec.Decode(payload.ToByteArray()));
    }

    /// <summary>
    ///     A birth metric declaring name/alias/datatype and <b>no value</b> — a birth's job here is to
    ///     declare the catalog, and these tests assert on the catalog, never on a published value.
    /// </summary>
    private static Payload.Types.Metric SpBirthMetric(string name, ulong alias, TahuDataType type)
        => new() { Name = name, Alias = alias, Datatype = (uint)type };

    // ---- test doubles ----

    /// <summary>
    ///     Records the streamed discovery tree instead of materializing OPC UA nodes. Local to this
    ///     suite: the driver test project does not reference <c>Commons</c>, where the runtime's own
    ///     capturing builder lives. Mirrors the per-driver <c>RecordingBuilder</c> the AB CIP / FOCAS
    ///     suites use.
    /// </summary>
    private sealed class CapturingAddressSpaceBuilder : IAddressSpaceBuilder
    {
        /// <summary>Every folder streamed, in order, across the whole tree.</summary>
        public List<(string BrowseName, string DisplayName)> Folders { get; } = [];

        /// <summary>Every variable streamed, in order, across the whole tree.</summary>
        public List<(string BrowseName, DriverAttributeInfo Info)> Variables { get; } = [];

        /// <inheritdoc />
        public IAddressSpaceBuilder Folder(string browseName, string displayName)
        {
            Folders.Add((browseName, displayName));
            return this;
        }

        /// <inheritdoc />
        public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
        {
            Variables.Add((browseName, attributeInfo));
            return new Handle(attributeInfo.FullName);
        }

        /// <inheritdoc />
        public void AddProperty(string browseName, DriverDataType dataType, object? value) { }

        private sealed class Handle(string fullRef) : IVariableHandle
        {
            public string FullReference => fullRef;

            public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new NullSink();
        }

        private sealed class NullSink : IAlarmConditionSink
        {
            public void OnTransition(AlarmEventArgs args) { }
        }
    }
}