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>
///     Covers <see cref="SparkplugIngestor"/> (Task 21) — the design doc's §3.6 correctness matrix:
///     rebirth, missed birth, seq gap, alias reuse across a rebirth, death → stale → rebirth, plus
///     data-before-birth and unknown-alias. Every test runs without a broker: payloads are built with
///     the generated Tahu types, decoded through the real <see cref="SparkplugCodec"/>, and fed to
///     <see cref="SparkplugIngestor.Dispatch"/> (the method <see cref="SparkplugIngestor.HandleMessage"/>
///     calls on MQTTnet's dispatcher thread).
/// </summary>
/// <remarks>
///     <b>The identity assertion is the point of half of these.</b> The plan's illustrative snippet
///     asserts <c>firedRef.ShouldContain("Filler1:Temperature")</c> — the retired pre-v3 shape. A
///     Sparkplug-composed reference satisfies it and is silently dead against <c>DriverHostActor</c>'s
///     RawPath-keyed dual-namespace fan-out, so every publish here is asserted to <b>be</b> the RawPath
///     and to <b>not</b> be the composed form.
/// </remarks>
public sealed class SparkplugIngestorTests
{
    // OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled.
    private const uint Good = 0x00000000u;
    private const uint BadCommunicationError = 0x80050000u;
    private const uint BadTypeMismatch = 0x80740000u;
    private const uint BadNodeIdUnknown = 0x80340000u;

    private const string Group = "Plant1";
    private const string Node = "EdgeA";
    private const string Device = "Filler1";

    private const string TempPath = "Plant/Mqtt/spb/Filler1Temp";
    private const string PressPath = "Plant/Mqtt/spb/Filler1Press";
    private const string CountPath = "Plant/Mqtt/spb/Filler1Count";
    private const string NodeUptimePath = "Plant/Mqtt/spb/EdgeAUptime";

    /// <summary>The retired pre-v3 composed shape — asserted ABSENT, never present.</summary>
    private const string ComposedRef = "Plant1/EdgeA/Filler1:Temperature";

    private const ulong TempAlias = 5UL;
    private const ulong PressAlias = 6UL;
    private const ulong CountAlias = 9UL;

    // ---------------------------------------------------------------------------------
    // §3.6 #1/#2 — birth → data, resolved BY ALIAS, published UNDER THE RAWPATH
    // ---------------------------------------------------------------------------------

    /// <summary>
    ///     The plan's first snippet, corrected: the published reference is asserted to BE the tag's
    ///     RawPath, not merely to contain the composed <c>Filler1:Temperature</c> form.
    /// </summary>
    [Fact]
    public async Task BirthThenData_ResolvesByAlias_RaisesOnDataChangeUnderRawPath()
    {
        var ing = await NewSubscribedIngestorAsync();

        string? firedRef = null;
        object? firedValue = null;
        ing.OnDataChange += (_, e) => { firedRef = e.FullReference; firedValue = e.Snapshot.Value; };

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1));
        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f)));

        firedRef.ShouldBe(TempPath);
        firedRef.ShouldNotBe(ComposedRef);
        firedValue.ShouldBe(21.5f);
    }

    /// <summary>
    ///     Falsifiability control for the identity assertion above: the composed Sparkplug reference —
    ///     the shape the plan's own snippet would have accepted — must never be published.
    /// </summary>
    [Fact]
    public async Task PublishedReference_IsNeverTheComposedSparkplugForm()
    {
        var ing = await NewSubscribedIngestorAsync();
        var refs = new List<string>();
        ing.OnDataChange += (_, e) => refs.Add(e.FullReference);

        FeedBirths(ing);
        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f)));

        refs.ShouldNotBeEmpty();
        refs.ShouldAllBe(r => r == TempPath || r == PressPath || r == CountPath || r == NodeUptimePath);
        refs.ShouldNotContain(ComposedRef);
    }

    /// <summary>A DATA metric carries an alias and NO name — the shape every real Sparkplug DATA has.</summary>
    [Fact]
    public async Task DataMetric_CarriesNoNameOrDatatype_StillResolves()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var payload = Ddata(seq: 2, (TempAlias, TahuDataType.Float, 33.5f));
        payload.Metrics.ShouldAllBe(m => m.Name == null && m.DataType == null);

        var seen = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot;
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), payload);

        seen[TempPath].Value.ShouldBe(33.5f);
        seen[TempPath].StatusCode.ShouldBe(Good);
    }

    /// <summary>A node-scoped metric (NDATA) binds against the node's own scope, not a device's.</summary>
    [Fact]
    public async Task NodeScopedMetric_BindsToTheNodeScope()
    {
        var ing = await NewSubscribedIngestorAsync();
        var seen = new Dictionary<string, object?>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value;

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.NDATA), Ndata(seq: 1, (7UL, TahuDataType.Int64, 99L)));

        seen[NodeUptimePath].ShouldBe(99L);
        seen.ShouldNotContainKey(TempPath);
    }

    // ---------------------------------------------------------------------------------
    // §3.6 #1 — ALIAS REUSE ACROSS A REBIRTH (the mis-route this whole design prevents)
    // ---------------------------------------------------------------------------------

    /// <summary>
    ///     The single most important test in the file. An edge node is free to point alias 5 at a
    ///     different metric after a rebirth. Binding a tag to an alias silently routes Pressure into the
    ///     Temperature tag — good quality, plausible value, completely wrong.
    /// </summary>
    [Fact]
    public async Task AliasReusedAcrossRebirth_RoutesByName_NotByAlias()
    {
        var ing = await NewSubscribedIngestorAsync();
        var seen = new Dictionary<string, object?>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value;

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1));
        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f)));
        seen[TempPath].ShouldBe(21.5f);

        // Rebirth: the SAME alias 5 now names Pressure, and Temperature moves to alias 6.
        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2));
        ing.Dispatch(
            Topic(SparkplugMessageType.DBIRTH, Device),
            DbirthCustom(
                seq: 1,
                ("Pressure", TempAlias, TahuDataType.Float, 1.0f),
                ("Temperature", PressAlias, TahuDataType.Float, 2.0f)));

        seen.Clear();
        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 2, (TempAlias, TahuDataType.Float, 4.25f)));

        // Alias 5 is Pressure NOW. A tag bound by alias would put 4.25 on the Temperature RawPath.
        seen.ShouldContainKey(PressPath);
        seen[PressPath].ShouldBe(4.25f);
        seen.ShouldNotContainKey(TempPath);
    }

    /// <summary>
    ///     A rebirth REPLACES the catalog: an alias the new birth does not declare must stop resolving,
    ///     rather than lingering from the previous birth's table.
    /// </summary>
    [Fact]
    public async Task Rebirth_DroppingAMetric_StopsFeedingIt()
    {
        var ing = await NewSubscribedIngestorAsync();
        var seen = new List<string>();
        ing.OnDataChange += (_, e) => seen.Add(e.FullReference);

        FeedBirths(ing);
        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2));
        ing.Dispatch(
            Topic(SparkplugMessageType.DBIRTH, Device),
            DbirthCustom(seq: 1, ("Pressure", PressAlias, TahuDataType.Float, 0f)));

        seen.Clear();
        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f)));

        seen.ShouldNotContain(TempPath);
    }

    // ---------------------------------------------------------------------------------
    // §3.6 #3 — data before birth / unknown alias / seq gap ⇒ rebirth NCMD
    // ---------------------------------------------------------------------------------

    /// <summary>The plan's second snippet, against the real NCMD publish seam.</summary>
    [Fact]
    public async Task DataBeforeBirth_RequestsRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);

        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
    }

    /// <summary>An alias the current birth never declared is our metadata being behind the node's.</summary>
    [Fact]
    public async Task UnknownAlias_RequestsRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);
        FeedBirths(ing);
        publish.Topics.Clear();

        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 2, (4242UL, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
    }

    /// <summary>A missed message in the edge node's stream — §3.6 #3.</summary>
    [Fact]
    public async Task SeqGap_RequestsRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);
        FeedBirths(ing);
        publish.Topics.Clear();

        // seq 2 was expected; 4 means two messages were lost.
        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 4, (TempAlias, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
    }

    /// <summary>
    ///     Falsifiability control for every rebirth test above: a contiguous, birth-synchronised stream
    ///     must publish NOTHING. Without this the rebirth assertions could pass off a driver that
    ///     NCMDs on every single message.
    /// </summary>
    [Fact]
    public async Task ContiguousStream_RequestsNoRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);

        FeedBirths(ing);
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f)));
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 3, (TempAlias, TahuDataType.Float, 2f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldBeEmpty();
    }

    /// <summary>
    ///     255 → 0 is the sequence continuing, not a gap. Reading the wrap as a gap asks every edge node
    ///     to rebirth once per 256 messages, so a busy node spends its life republishing births.
    /// </summary>
    [Fact]
    public async Task SeqWrapsFrom255To0_IsNotAGap()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 255, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 0));
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 1, (TempAlias, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldBeEmpty();
    }

    /// <summary>
    ///     A data message whose sequence happens to be contiguous but which arrived with no NBIRTH seen
    ///     since connect — the driver joined mid-stream and holds no alias table.
    ///     <see cref="SequenceTracker.Accept"/> returns <see langword="true"/> for it, so a driver that
    ///     only read the return value would sit happily behind a stream it cannot decode.
    /// </summary>
    [Fact]
    public async Task LateJoin_FirstMessageIsNotABirth_RequestsRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);

        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 17));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
    }

    /// <summary>
    ///     Repeatedly commanding a node that is already republishing IS the storm the design warns
    ///     about. Three unroutable messages in a row must produce one NCMD, not three.
    /// </summary>
    [Fact]
    public async Task RepeatedRebirthTriggers_AreDebouncedPerEdgeNode()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish, rebirthDebounce: TimeSpan.FromMinutes(5));

        for (var i = 0; i < 3; i++)
        {
            ing.Dispatch(
                Topic(SparkplugMessageType.DDATA, Device),
                Ddata(seq: (ulong)i, (CountAlias, TahuDataType.Float, 1f)));
        }

        await ing.DrainPendingRebirthsAsync();

        publish.Topics.Count.ShouldBe(1);
    }

    /// <summary>The debounce floor is per edge node, not global — a second node must not be starved.</summary>
    [Fact]
    public async Task RebirthDebounce_IsPerEdgeNode()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(
            publish,
            rebirthDebounce: TimeSpan.FromMinutes(5),
            extraTags: [Tag("Plant/Mqtt/spb/EdgeBTemp", Blob(Group, "EdgeB", Device, "Temperature", "Float32"))]);

        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f)));
        ing.Dispatch(
            new SparkplugTopic(SparkplugMessageType.DDATA, Group, "EdgeB", Device, null),
            Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeB");
    }

    /// <summary><c>requestRebirthOnGap = false</c> is the operator's off switch for ALL outbound NCMDs.</summary>
    [Fact]
    public async Task RequestRebirthOnGapDisabled_PublishesNothing()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish, requestRebirthOnGap: false);

        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldBeEmpty();
    }

    /// <summary>A missing publish transport degrades to a log, never an exception on the dispatcher.</summary>
    [Fact]
    public async Task RebirthWithNoPublishTransport_DoesNotThrow()
    {
        var ing = await NewSubscribedIngestorAsync(publish: null);

        Should.NotThrow(() => ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f))));
    }

    // ---------------------------------------------------------------------------------
    // §3.6 #4 — death → STALE, next birth restores Good
    // ---------------------------------------------------------------------------------

    /// <summary>The plan's third snippet.</summary>
    [Fact]
    public async Task Ddeath_EmitsBadForDeviceMetrics()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var quals = new List<uint>();
        ing.OnDataChange += (_, e) => quals.Add(e.Snapshot.StatusCode);

        ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2));

        quals.ShouldNotBeEmpty();
        quals.ShouldAllBe(q => q == BadCommunicationError);
    }

    /// <summary>The completing half of §3.6 #4 — a birth is a full snapshot, so it restores Good.</summary>
    [Fact]
    public async Task DdeathThenDbirth_RestoresGood()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var last = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot;

        ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2));
        last[TempPath].StatusCode.ShouldBe(BadCommunicationError);

        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 3));

        last[TempPath].StatusCode.ShouldBe(Good);
        last[TempPath].Value.ShouldBe(0f);
    }

    /// <summary>A DDEATH stales exactly its own device; the node's own metrics keep flowing.</summary>
    [Fact]
    public async Task Ddeath_DoesNotStaleTheNodesOwnMetrics()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var last = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot;

        ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2));

        last.ShouldContainKey(TempPath);
        last.ShouldNotContainKey(NodeUptimePath);
    }

    /// <summary>An NDEATH kills the node AND every device under it.</summary>
    [Fact]
    public async Task Ndeath_StalesTheNodeAndEveryDeviceUnderIt()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var last = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot;

        ing.Dispatch(Topic(SparkplugMessageType.NDEATH), Ndeath(bdSeq: 1));

        last[TempPath].StatusCode.ShouldBe(BadCommunicationError);
        last[NodeUptimePath].StatusCode.ShouldBe(BadCommunicationError);
    }

    /// <summary>
    ///     The <c>bdSeq</c> tie of §3.6 #3. A node's connection drops, it reconnects and re-births, and
    ///     only THEN does the broker deliver the old session's Last Will. Acting on it marks a live node
    ///     dead with nothing coming to correct it.
    /// </summary>
    [Fact]
    public async Task StaleNdeath_FromAPreviousSession_IsIgnored()
    {
        var ing = await NewSubscribedIngestorAsync();

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1));

        var quals = new List<uint>();
        ing.OnDataChange += (_, e) => quals.Add(e.Snapshot.StatusCode);

        ing.Dispatch(Topic(SparkplugMessageType.NDEATH), Ndeath(bdSeq: 1));

        quals.ShouldBeEmpty();
        ing.Births.Find(new SparkplugScope(Group, Node, Device)).ShouldNotBeNull();
    }

    /// <summary>
    ///     <b>NDEATH must never reach <see cref="SequenceTracker.Accept"/>.</b> It is the broker-published
    ///     Last Will and carries no <c>seq</c> at all, so <c>Accept</c> would report a gap on every death
    ///     and answer a node that has just died with a rebirth request it can never serve.
    /// </summary>
    [Fact]
    public async Task Ndeath_DoesNotAdvanceTheSequenceTracker_AndRequestsNoRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);

        FeedBirths(ing);
        publish.Topics.Clear();

        ing.Dispatch(Topic(SparkplugMessageType.NDEATH), Ndeath(bdSeq: 1));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldBeEmpty();
    }

    /// <summary>
    ///     Per the Sparkplug spec an NBIRTH voids every prior DBIRTH under that node: those devices MUST
    ///     re-DBIRTH, and until they do their aliases mean nothing.
    /// </summary>
    [Fact]
    public async Task Nbirth_InvalidatesPriorDeviceBirths_AndStalesThem()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var last = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot;

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2));

        last[TempPath].StatusCode.ShouldBe(BadCommunicationError);
        ing.Births.Find(new SparkplugScope(Group, Node, Device)).ShouldBeNull();
    }

    // ---------------------------------------------------------------------------------
    // Wire-level value correctness
    // ---------------------------------------------------------------------------------

    /// <summary>
    ///     Sparkplug carries Int8/16/32/64 as two's complement in an <b>unsigned</b> proto field, so a
    ///     metric whose value is -43 arrives as 4294967253. Skipping
    ///     <see cref="SparkplugMetricBinding.Reinterpret"/> publishes a plausible, Good-quality,
    ///     completely wrong number — or, once the tag is Int32, an out-of-range refusal.
    /// </summary>
    [Fact]
    public async Task NegativeInteger_IsReinterpretedFromItsTwosComplementWireValue()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var seen = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot;

        ing.Dispatch(
            Topic(SparkplugMessageType.DDATA, Device),
            Ddata(seq: 2, (CountAlias, TahuDataType.Int32, -43)));

        seen[CountPath].StatusCode.ShouldBe(Good);
        seen[CountPath].Value.ShouldBe(-43);
        seen[CountPath].Value.ShouldNotBe(4294967253L);
    }

    /// <summary>The birth's own values are published too — that is how a rebirth restores Good.</summary>
    [Fact]
    public async Task BirthValues_ArePublished()
    {
        var ing = await NewSubscribedIngestorAsync();
        var seen = new Dictionary<string, object?>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value;

        FeedBirths(ing);

        seen[TempPath].ShouldBe(0f);
        seen[CountPath].ShouldBe(-42);
    }

    /// <summary>
    ///     A tag authored with no <c>dataType</c> takes the type its BIRTH declared — the reason
    ///     <c>dataType</c> is optional in the Sparkplug tag shape at all.
    /// </summary>
    [Fact]
    public async Task TagWithNoAuthoredDataType_TakesTheBirthsType()
    {
        const string Untyped = "Plant/Mqtt/spb/Untyped";
        var ing = NewIngestor(tags: [Tag(Untyped, """{"groupId":"Plant1","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"Temperature"}""")]);
        await ing.SubscribeAsync([Untyped], TimeSpan.Zero, TestContext.Current.CancellationToken);

        object? value = null;
        ing.OnDataChange += (_, e) => value = e.Snapshot.Value;

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7.5f)));

        // Float32, from the birth — NOT the record's String default, which would have produced "7.5".
        value.ShouldBe(7.5f);
    }

    /// <summary>An authored <c>dataType</c> wins over the birth's — the operator declared it deliberately.</summary>
    [Fact]
    public async Task AuthoredDataType_WinsOverTheBirths()
    {
        const string AsText = "Plant/Mqtt/spb/TempAsText";
        var ing = NewIngestor(tags: [Tag(AsText, Blob(Group, Node, Device, "Temperature", "String"))]);
        await ing.SubscribeAsync([AsText], TimeSpan.Zero, TestContext.Current.CancellationToken);

        object? value = null;
        ing.OnDataChange += (_, e) => value = e.Snapshot.Value;

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7.5f)));

        value.ShouldBe("7.5");
    }

    /// <summary>An explicit Sparkplug <c>is_null</c> is not a value of the tag's declared type.</summary>
    [Fact]
    public async Task ExplicitNullMetric_PublishesBadTypeMismatch()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var seen = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot;

        var payload = new Payload { Seq = 2UL, Timestamp = 1UL };
        payload.Metrics.Add(new Payload.Types.Metric { Alias = TempAlias, IsNull = true });
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(payload));

        seen[TempPath].StatusCode.ShouldBe(BadTypeMismatch);
        seen[TempPath].Value.ShouldBeNull();
    }

    /// <summary>
    ///     A DataSet/Template metric is warn-and-skip — NOT a missing metric and NOT a rebirth trigger.
    ///     A node asked to rebirth would answer with the same unsupported metric and the pair would loop.
    /// </summary>
    [Fact]
    public async Task UnsupportedMetricKind_IsSkipped_AndRequestsNoRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);
        FeedBirths(ing);
        publish.Topics.Clear();

        var seen = new List<string>();
        ing.OnDataChange += (_, e) => seen.Add(e.FullReference);

        var payload = new Payload { Seq = 2UL, Timestamp = 1UL };
        payload.Metrics.Add(new Payload.Types.Metric { Alias = TempAlias, DatasetValue = new Payload.Types.DataSet() });
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(payload));
        await ing.DrainPendingRebirthsAsync();

        seen.ShouldBeEmpty();
        publish.Topics.ShouldBeEmpty();
    }

    /// <summary>A value that does not fit the authored type is refused, never rounded or wrapped.</summary>
    [Fact]
    public async Task FractionalValueForAnIntegerTag_PublishesBadTypeMismatch()
    {
        const string IntTag = "Plant/Mqtt/spb/TempAsInt";
        var ing = NewIngestor(tags: [Tag(IntTag, Blob(Group, Node, Device, "Temperature", "Int32"))]);
        await ing.SubscribeAsync([IntTag], TimeSpan.Zero, TestContext.Current.CancellationToken);

        DataValueSnapshot? snapshot = null;
        ing.OnDataChange += (_, e) => snapshot = e.Snapshot;

        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f)));

        snapshot!.StatusCode.ShouldBe(BadTypeMismatch);
    }

    // ---------------------------------------------------------------------------------
    // Robustness: nothing may throw or mutate state off bad input
    // ---------------------------------------------------------------------------------

    /// <summary>
    ///     An undecodable body is not evidence of anything. It must not clear an alias table, advance a
    ///     sequence tracker, or evict a birth — a hostile publisher on a group-wide <c>#</c> subscription
    ///     would otherwise be able to blank the address space with garbage bytes.
    /// </summary>
    [Fact]
    public async Task InvalidPayload_MutatesNothing()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);
        FeedBirths(ing);
        publish.Topics.Clear();

        ing.HandleMessage("spBv1.0/Plant1/DDATA/EdgeA/Filler1", [0xDE, 0xAD, 0xBE, 0xEF], retained: false);
        await ing.DrainPendingRebirthsAsync();

        // The birth survived …
        ing.Births.Find(new SparkplugScope(Group, Node, Device)).ShouldNotBeNull();

        // … the sequence baseline survived (seq 2 is still contiguous, so no rebirth) …
        var seen = new Dictionary<string, object?>(StringComparer.Ordinal);
        ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value;
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 9f)));
        await ing.DrainPendingRebirthsAsync();

        seen[TempPath].ShouldBe(9f);
        publish.Topics.ShouldBeEmpty();
    }

    /// <summary>An undecodable NDEATH must not kill a live node either.</summary>
    [Fact]
    public async Task InvalidNdeathPayload_DoesNotStale()
    {
        var ing = await NewSubscribedIngestorAsync();
        FeedBirths(ing);

        var quals = new List<uint>();
        ing.OnDataChange += (_, e) => quals.Add(e.Snapshot.StatusCode);

        ing.HandleMessage("spBv1.0/Plant1/NDEATH/EdgeA", [0x01, 0x02, 0x03], retained: false);

        quals.ShouldBeEmpty();
    }

    /// <summary>
    ///     Runs on MQTTnet's dispatcher thread: an escaping exception stalls delivery for every
    ///     subscription on the connection.
    /// </summary>
    [Theory]
    [InlineData("")]
    [InlineData("not/a/sparkplug/topic")]
    [InlineData("spBv1.0/Plant1/NBIRTH/EdgeA")]
    [InlineData("spBv1.0/Plant1/DDATA/EdgeA/Filler1")]
    [InlineData("spBv1.0/STATE/otopcua-host-1")]
    [InlineData("spBv1.0")]
    public async Task HandleMessage_NeverThrows(string topic)
    {
        var ing = await NewSubscribedIngestorAsync();

        Should.NotThrow(() => ing.HandleMessage(topic, [], retained: false));
        Should.NotThrow(() => ing.HandleMessage(topic, [0xFF, 0x00, 0x7F], retained: true));
    }

    /// <summary>A throwing subscriber is contained rather than escaping onto the dispatcher thread.</summary>
    [Fact]
    public async Task ThrowingSubscriber_IsContained()
    {
        var ing = await NewSubscribedIngestorAsync();
        ing.OnDataChange += (_, _) => throw new InvalidOperationException("boom");

        Should.NotThrow(() => FeedBirths(ing));
    }

    /// <summary>An oversize body is refused BEFORE any protobuf parse, and mutates nothing.</summary>
    [Fact]
    public async Task OversizePayload_IsDroppedBeforeDecoding()
    {
        var ing = NewIngestor(maxPayloadBytes: 16);
        await ing.SubscribeAsync([TempPath], TimeSpan.Zero, TestContext.Current.CancellationToken);

        var bytes = Nbirth(seq: 0, bdSeq: 1);
        bytes.Metrics.Count.ShouldBeGreaterThan(0);

        ing.HandleMessage("spBv1.0/Plant1/NBIRTH/EdgeA", new byte[64], retained: false);

        ing.Births.Count.ShouldBe(0);
    }

    /// <summary>
    ///     The driver subscribes <c>spBv1.0/{group}/#</c>, so a large plant delivers traffic for every
    ///     node in the group. Only authored edge nodes are processed — that is what bounds the tracker
    ///     and birth maps by configuration rather than by whatever the plant publishes.
    /// </summary>
    [Fact]
    public async Task UnauthoredEdgeNode_IsIgnoredEntirely()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);

        var fired = false;
        ing.OnDataChange += (_, _) => fired = true;

        ing.Dispatch(new SparkplugTopic(SparkplugMessageType.NBIRTH, Group, "StrangerNode", null, null), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(
            new SparkplugTopic(SparkplugMessageType.DDATA, Group, "StrangerNode", "Dev", null),
            Ddata(seq: 1, (TempAlias, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        fired.ShouldBeFalse();
        ing.Births.Count.ShouldBe(0);
        publish.Topics.ShouldBeEmpty();
    }

    /// <summary>
    ///     An unauthored DEVICE under an authored node still advances its edge node's sequence — dropping
    ///     it early would manufacture a gap on the very next authored message — and, because nothing in
    ///     that scope could ever be published, it must not demand a rebirth either. A spec-violating
    ///     device that publishes DDATA without ever DBIRTHing would otherwise drive a permanent NCMD at
    ///     the debounce cadence, forever.
    /// </summary>
    [Fact]
    public async Task UnauthoredDeviceUnderAnAuthoredNode_StillAdvancesTheSequence()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish);
        FeedBirths(ing);
        publish.Topics.Clear();

        // seq 2 belongs to a device nobody authored …
        ing.Dispatch(
            new SparkplugTopic(SparkplugMessageType.DDATA, Group, Node, "OtherDevice", null),
            Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f)));

        // … and seq 3 must therefore still be contiguous.
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 3, (TempAlias, TahuDataType.Float, 2f)));
        await ing.DrainPendingRebirthsAsync();

        publish.Topics.ShouldBeEmpty();
    }

    // ---------------------------------------------------------------------------------
    // Registration / resolution / read
    // ---------------------------------------------------------------------------------

    [Fact]
    public void Register_MapsOnlySparkplugShapedBlobs()
    {
        var ing = NewIngestor(tags: []);

        var mapped = ing.Register([
            Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32")),
            Tag("Plant/Mqtt/spb/Plainish", """{"topic":"f/oven/temp","payloadFormat":"Scalar","dataType":"Float64"}"""),
            Tag("Plant/Mqtt/spb/NoMetric", """{"groupId":"Plant1","edgeNodeId":"EdgeA"}"""),
        ]);

        mapped.ShouldBe(1);
        ing.TryResolve(TempPath, out var def).ShouldBeTrue();
        def.MetricName.ShouldBe("Temperature");
        def.Name.ShouldBe(TempPath);
    }

    [Fact]
    public void Register_IsWholesale_ADroppedTagStopsResolving()
    {
        var ing = NewIngestor();
        ing.TryResolve(TempPath, out _).ShouldBeTrue();

        ing.Register([]);

        ing.TryResolve(TempPath, out _).ShouldBeFalse();
    }

    [Fact]
    public async Task SubscribeAsync_UnauthoredReference_ReportsBadNodeIdUnknown()
    {
        var ing = NewIngestor();

        await ing.SubscribeAsync(["Plant/Mqtt/spb/NotAThing"], TimeSpan.Zero, TestContext.Current.CancellationToken);

        ing.Values.Read("Plant/Mqtt/spb/NotAThing").StatusCode.ShouldBe(BadNodeIdUnknown);
    }

    [Fact]
    public async Task UnsubscribedButAuthoredTag_IsCachedButNotRaised()
    {
        var ing = NewIngestor();
        var fired = false;
        ing.OnDataChange += (_, _) => fired = true;

        FeedBirths(ing);

        fired.ShouldBeFalse();
        ing.Values.Read(TempPath).Value.ShouldBe(0f);
    }

    /// <summary>Two raw tags may legitimately project the same Sparkplug metric; both must be fed.</summary>
    [Fact]
    public async Task TwoTagsBoundToOneMetric_BothReceive()
    {
        const string Mirror = "Plant/Mqtt/spb/Filler1TempMirror";
        var ing = NewIngestor(extraTags: [Tag(Mirror, Blob(Group, Node, Device, "Temperature", "Float32"))]);
        await ing.SubscribeAsync([TempPath, Mirror], TimeSpan.Zero, TestContext.Current.CancellationToken);

        var seen = new List<string>();
        ing.OnDataChange += (_, e) => seen.Add(e.FullReference);

        FeedBirths(ing);
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f)));

        seen.ShouldContain(TempPath);
        seen.ShouldContain(Mirror);
    }

    // ---------------------------------------------------------------------------------
    // Subscribe / reconnect
    // ---------------------------------------------------------------------------------

    [Fact]
    public async Task EstablishAsync_SubscribesTheGroupAndStateFilters_AndRequestsLateJoinRebirth()
    {
        var subscribe = new FakeSubscribeTransport();
        var publish = new RecordingPublishTransport();
        var ing = NewIngestor(publish, subscribe, hostId: "otopcua-host-1");

        await ing.EstablishAsync(TestContext.Current.CancellationToken);
        await ing.DrainPendingRebirthsAsync();

        subscribe.Filters.Select(f => f.Topic).ShouldBe(["spBv1.0/Plant1/#", "spBv1.0/STATE/otopcua-host-1"]);
        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
    }

    /// <summary>No host id ⇒ no STATE filter. This driver has no use for a stranger's host state.</summary>
    [Fact]
    public async Task EstablishAsync_WithNoHostId_SubscribesTheGroupFilterOnly()
    {
        var subscribe = new FakeSubscribeTransport();
        var ing = NewIngestor(subscribe: subscribe, hostId: "");

        await ing.EstablishAsync(TestContext.Current.CancellationToken);

        subscribe.Filters.Select(f => f.Topic).ShouldBe(["spBv1.0/Plant1/#"]);
    }

    /// <summary>
    ///     There is exactly one filter and it covers every tag, so a broker that refuses it leaves the
    ///     whole driver deaf — the one outcome that must be a visible fault rather than a healthy-looking
    ///     driver receiving nothing.
    /// </summary>
    [Fact]
    public async Task EstablishAsync_GroupFilterRefused_Throws()
    {
        var subscribe = new FakeSubscribeTransport { Reject = _ => true };
        var ing = NewIngestor(subscribe: subscribe);

        await Should.ThrowAsync<InvalidOperationException>(
            () => ing.EstablishAsync(TestContext.Current.CancellationToken));
    }

    /// <summary>A refused STATE filter costs an observability signal, not the data plane.</summary>
    [Fact]
    public async Task EstablishAsync_StateFilterRefused_DoesNotThrow()
    {
        var subscribe = new FakeSubscribeTransport { Reject = t => t.Contains("STATE", StringComparison.Ordinal) };
        var ing = NewIngestor(subscribe: subscribe, hostId: "otopcua-host-1");

        await ing.EstablishAsync(TestContext.Current.CancellationToken);
    }

    /// <summary>
    ///     Across an outage of unknown length an edge node may have restarted and rebound its aliases; a
    ///     DDATA resolved against the pre-outage table would route a value to whichever tag used to own
    ///     that alias.
    /// </summary>
    [Fact]
    public async Task OnReconnected_ClearsBirths_ResubscribesAndRequestsRebirth()
    {
        var subscribe = new FakeSubscribeTransport();
        var publish = new RecordingPublishTransport();
        var ing = NewIngestor(publish, subscribe);
        await ing.SubscribeAsync([TempPath], TimeSpan.Zero, TestContext.Current.CancellationToken);

        FeedBirths(ing);
        ing.Births.Count.ShouldBeGreaterThan(0);
        publish.Topics.Clear();
        subscribe.Filters.Clear();

        await ing.OnReconnectedAsync(TestContext.Current.CancellationToken);
        await ing.DrainPendingRebirthsAsync();

        ing.Births.Count.ShouldBe(0);
        subscribe.Filters.Select(f => f.Topic).ShouldContain("spBv1.0/Plant1/#");
        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
    }

    /// <summary>
    ///     Reconnect is a driver-side event, not a statement about the plant: last-known values remain
    ///     the best available answer until the rebirth lands. Only a death stales a tag.
    /// </summary>
    [Fact]
    public async Task OnReconnected_DoesNotStaleValues()
    {
        var ing = await NewSubscribedIngestorAsync(new RecordingPublishTransport(), new FakeSubscribeTransport());
        FeedBirths(ing);

        await ing.OnReconnectedAsync(TestContext.Current.CancellationToken);

        ing.Values.Read(TempPath).StatusCode.ShouldBe(Good);
    }

    /// <summary>After a reconnect the driver has no birth, so the first DDATA must ask for one.</summary>
    [Fact]
    public async Task AfterReconnect_DataResolvesAgainstNothing_AndRequestsRebirth()
    {
        var publish = new RecordingPublishTransport();
        var ing = await NewSubscribedIngestorAsync(publish, new FakeSubscribeTransport(), rebirthDebounce: TimeSpan.Zero);
        FeedBirths(ing);

        await ing.OnReconnectedAsync(TestContext.Current.CancellationToken);
        publish.Topics.Clear();

        var fired = false;
        ing.OnDataChange += (_, _) => fired = true;
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f)));
        await ing.DrainPendingRebirthsAsync();

        fired.ShouldBeFalse();
        publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA");
    }

    // ---------------------------------------------------------------------------------
    // STATE (receive only — this driver never publishes one; see the ingestor's remarks)
    // ---------------------------------------------------------------------------------

    [Fact]
    public async Task State_V30JsonPayload_IsObserved()
    {
        var ing = await NewSubscribedIngestorAsync();

        ing.HandleMessage("spBv1.0/STATE/otopcua-host-1", """{"online":true,"timestamp":1721822400000}"""u8, retained: true);

        ing.HostState!.HostId.ShouldBe("otopcua-host-1");
        ing.HostState!.Online.ShouldBeTrue();
    }

    [Fact]
    public async Task State_LegacyTextPayload_IsObserved()
    {
        var ing = await NewSubscribedIngestorAsync();

        ing.HandleMessage("STATE/otopcua-host-1", "OFFLINE"u8, retained: true);

        ing.HostState!.Online.ShouldBeFalse();
    }

    [Fact]
    public async Task State_UnrecognisedPayload_IsIgnored()
    {
        var ing = await NewSubscribedIngestorAsync();

        ing.HandleMessage("spBv1.0/STATE/otopcua-host-1", "maybe?"u8, retained: true);

        ing.HostState.ShouldBeNull();
    }

    // ---------------------------------------------------------------------------------
    // Task 22 seam
    // ---------------------------------------------------------------------------------

    /// <summary>
    ///     The rediscovery seam Task 22 wires to <c>IRediscoverable.OnRediscoveryNeeded</c>. Deliberately
    ///     unwired in <c>MqttDriver</c> today — this asserts the event exists and carries the scope +
    ///     metric set that Task 22 needs to decide whether the tree changed.
    /// </summary>
    [Fact]
    public async Task BirthObserved_FiresWithTheScopeAndMetricNames()
    {
        var ing = await NewSubscribedIngestorAsync();
        var observed = new List<SparkplugBirthObservedEventArgs>();
        ing.BirthObserved += (_, e) => observed.Add(e);

        FeedBirths(ing);

        observed.Count.ShouldBe(2);
        observed[0].Scope.ShouldBe(new SparkplugScope(Group, Node, null));
        observed[1].Scope.ShouldBe(new SparkplugScope(Group, Node, Device));
        observed[1].MetricNames.ShouldContain("Temperature");
    }

    // =================================================================================
    // Fixtures
    // =================================================================================

    private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false);

    private static string Blob(string group, string node, string? device, string metric, string? dataType)
    {
        var deviceKey = device is null ? "" : $"""{'"'}deviceId{'"'}:"{device}",""";
        var typeKey = dataType is null ? "" : $""","dataType":"{dataType}" """;
        return $$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}",{{deviceKey}}"metricName":"{{metric}}"{{typeKey}}}""";
    }

    /// <summary>The standard authored set: three device metrics plus one node-level metric.</summary>
    private static RawTagEntry[] DefaultTags() =>
    [
        Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32")),
        Tag(PressPath, Blob(Group, Node, Device, "Pressure", "Float32")),
        Tag(CountPath, Blob(Group, Node, Device, "Count", "Int32")),
        Tag(NodeUptimePath, Blob(Group, Node, null, "Uptime", "Int64")),
    ];

    private static SparkplugIngestor NewIngestor(
        RecordingPublishTransport? publish = null,
        FakeSubscribeTransport? subscribe = null,
        string hostId = "otopcua-host-1",
        bool requestRebirthOnGap = true,
        TimeSpan? rebirthDebounce = null,
        int maxPayloadBytes = MqttSubscriptionManager.DefaultMaxPayloadBytes,
        RawTagEntry[]? tags = null,
        RawTagEntry[]? extraTags = null)
    {
        var options = new MqttDriverOptions
        {
            Mode = MqttMode.SparkplugB,
            Sparkplug = new MqttSparkplugOptions
            {
                GroupId = Group,
                HostId = hostId,
                RequestRebirthOnGap = requestRebirthOnGap,
            },
        };

        var ingestor = new SparkplugIngestor(
            options,
            "mqtt-spb-1",
            subscribe,
            publish,
            logger: null,
            cache: null,
            maxPayloadBytes: maxPayloadBytes,
            // Zero by default so a test asserting "these two triggers each produce an NCMD" is not
            // silently satisfied by the production debounce; the debounce itself has its own tests.
            rebirthDebounce: rebirthDebounce ?? TimeSpan.Zero);

        ingestor.Register([.. tags ?? DefaultTags(), .. extraTags ?? []]);
        return ingestor;
    }

    /// <summary>An ingestor with every default tag registered AND subscribed, so notifications fire.</summary>
    private static async Task<SparkplugIngestor> NewSubscribedIngestorAsync(
        RecordingPublishTransport? publish = null,
        FakeSubscribeTransport? subscribe = null,
        bool requestRebirthOnGap = true,
        TimeSpan? rebirthDebounce = null,
        RawTagEntry[]? extraTags = null)
    {
        var ingestor = NewIngestor(
            publish,
            subscribe,
            requestRebirthOnGap: requestRebirthOnGap,
            rebirthDebounce: rebirthDebounce,
            extraTags: extraTags);

        await ingestor.SubscribeAsync(
            [.. ingestor.AuthoredRawPaths],
            TimeSpan.Zero,
            TestContext.Current.CancellationToken);

        return ingestor;
    }

    /// <summary>Feeds the standard NBIRTH (seq 0, bdSeq 1) + DBIRTH (seq 1) pair.</summary>
    private static void FeedBirths(SparkplugIngestor ing)
    {
        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1));
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1));
    }

    private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) =>
        new(type, Group, Node, device, null);

    private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray());

    /// <summary>An NBIRTH: the <c>bdSeq</c> session token plus the node's own metric catalog.</summary>
    private static SparkplugPayload Nbirth(ulong seq, ulong bdSeq)
    {
        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(BirthMetric("Uptime", 7UL, TahuDataType.Int64, 0L));
        return Decode(payload);
    }

    /// <summary>The standard DBIRTH: Temperature@5 (Float), Pressure@6 (Float), Count@9 (Int32, -42).</summary>
    private static SparkplugPayload Dbirth(ulong seq) => DbirthCustom(
        seq,
        ("Temperature", TempAlias, TahuDataType.Float, 0f),
        ("Pressure", PressAlias, TahuDataType.Float, 0f),
        ("Count", CountAlias, TahuDataType.Int32, -42));

    private static SparkplugPayload DbirthCustom(
        ulong seq,
        params (string Name, ulong Alias, TahuDataType Type, object Value)[] metrics)
    {
        var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
        foreach (var (name, alias, type, value) in metrics)
        {
            payload.Metrics.Add(BirthMetric(name, alias, type, value));
        }

        return Decode(payload);
    }

    /// <summary>A DDATA: metrics carry an alias and NOTHING else — the real post-birth wire shape.</summary>
    private static SparkplugPayload Ddata(ulong seq, params (ulong Alias, TahuDataType Type, object Value)[] metrics)
        => DataPayload(seq, metrics);

    /// <summary>An NDATA — identical shape to <see cref="Ddata"/>; the topic is what makes it node-scoped.</summary>
    private static SparkplugPayload Ndata(ulong seq, params (ulong Alias, TahuDataType Type, object Value)[] metrics)
        => DataPayload(seq, metrics);

    private static SparkplugPayload DataPayload(
        ulong seq,
        (ulong Alias, TahuDataType Type, object Value)[] metrics)
    {
        var payload = new Payload { Seq = seq, Timestamp = 1721822405000UL };
        foreach (var (alias, type, value) in metrics)
        {
            // Deliberately NO Name and NO Datatype: that is what a Sparkplug DATA metric looks like
            // after a birth, and it is the case an "absent means empty string" consumer gets wrong.
            var metric = new Payload.Types.Metric { Alias = alias };
            SetWireValue(metric, type, value);
            payload.Metrics.Add(metric);
        }

        return Decode(payload);
    }

    /// <summary>A DDEATH — published by the edge node, and therefore sequenced.</summary>
    private static SparkplugPayload Ddeath(ulong seq) =>
        Decode(new Payload { Seq = seq, Timestamp = 1721822405000UL });

    /// <summary>An NDEATH — the broker-published Last Will: a <c>bdSeq</c> and NO <c>seq</c> at all.</summary>
    private static SparkplugPayload Ndeath(ulong bdSeq)
    {
        var payload = new Payload { Timestamp = 1721822405000UL };
        payload.Metrics.Add(new Payload.Types.Metric
        {
            Name = "bdSeq",
            Datatype = (uint)TahuDataType.Uint64,
            LongValue = bdSeq,
        });
        return Decode(payload);
    }

    private static Payload.Types.Metric BirthMetric(string name, ulong alias, TahuDataType type, object value)
    {
        var metric = new Payload.Types.Metric { Name = name, Alias = alias, Datatype = (uint)type };
        SetWireValue(metric, type, value);
        return metric;
    }

    /// <summary>
    ///     Writes a value into the <c>oneof</c> field Sparkplug actually carries it in — including the
    ///     two's-complement-in-an-unsigned-field encoding for the signed integer types, which is the
    ///     whole reason <see cref="SparkplugMetricBinding.Reinterpret"/> exists.
    /// </summary>
    private static void SetWireValue(Payload.Types.Metric metric, TahuDataType type, object value)
    {
        switch (type)
        {
            case TahuDataType.Boolean:
                metric.BooleanValue = (bool)value;
                break;
            case TahuDataType.Float:
                metric.FloatValue = (float)value;
                break;
            case TahuDataType.Double:
                metric.DoubleValue = (double)value;
                break;
            case TahuDataType.Int8:
            case TahuDataType.Int16:
            case TahuDataType.Int32:
                metric.IntValue = unchecked((uint)Convert.ToInt32(value));
                break;
            case TahuDataType.Uint8:
            case TahuDataType.Uint16:
            case TahuDataType.Uint32:
                metric.IntValue = Convert.ToUInt32(value);
                break;
            case TahuDataType.Int64:
                metric.LongValue = unchecked((ulong)Convert.ToInt64(value));
                break;
            case TahuDataType.Uint64:
            case TahuDataType.DateTime:
                metric.LongValue = Convert.ToUInt64(value);
                break;
            case TahuDataType.String:
            case TahuDataType.Text:
            case TahuDataType.Uuid:
                metric.StringValue = (string)value;
                break;
            default:
                throw new NotSupportedException($"Test helper does not encode {type}.");
        }
    }

    /// <summary>Records every NCMD so the rebirth policy is assertable without a broker.</summary>
    private sealed class RecordingPublishTransport : IMqttPublishTransport
    {
        public List<string> Topics { get; } = [];

        public List<byte[]> Payloads { get; } = [];

        public Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken)
        {
            lock (Topics)
            {
                Topics.Add(topic);
                Payloads.Add(payload);
            }

            // A retained NCMD replays to the edge node on every future subscribe, driving a permanent
            // rebirth loop from one stale command. Assert it here so every rebirth path is covered.
            retain.ShouldBeFalse();
            return Task.CompletedTask;
        }
    }

    /// <summary>Records every SUBSCRIBE so the group/STATE filters are assertable without a broker.</summary>
    private sealed class FakeSubscribeTransport : IMqttSubscribeTransport
    {
        public List<MqttTopicSubscription> Filters { get; } = [];

        /// <summary>Topics for which the fake SUBACK reports a rejection.</summary>
        public Func<string, bool> Reject { get; set; } = _ => false;

        public Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
            IReadOnlyList<MqttTopicSubscription> filters,
            CancellationToken cancellationToken)
        {
            Filters.AddRange(filters);
            IReadOnlyList<MqttSubscribeOutcome> outcomes =
            [
                .. filters.Select(f => new MqttSubscribeOutcome(
                    f.Topic,
                    !Reject(f.Topic),
                    Reject(f.Topic) ? "NotAuthorized" : "GrantedQoS1")),
            ];
            return Task.FromResult(outcomes);
        }
    }
}