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

using System.Collections.Concurrent;
using System.Globalization;
using System.Text;
using Google.Protobuf;
using Microsoft.Extensions.Logging;
using MQTTnet;
using Org.Eclipse.Tahu.Protobuf;
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;

// See SparkplugTopicTests.cs for why this alias is redeclared locally in every consuming project
// rather than inherited: `SparkplugDataType` is a `global using` alias for the generated
// `Org.Eclipse.Tahu.Protobuf.DataType` (declared in Contracts/SparkplugDataType.cs), and `global using`
// scope does not cross a ProjectReference.
using SparkplugDataType = Org.Eclipse.Tahu.Protobuf.DataType;

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

/// <summary>
///     A passive MQTT <b>observation window</b>, served through the AdminUI's standard browse-session
///     interface.
///     <para>
///         MQTT has no discovery protocol — a broker cannot be asked "what topics exist". The only way
///         to learn the topic space is to subscribe to a wildcard and watch what arrives. So this
///         session accumulates every topic it observes into a segment tree (split on <c>/</c>) and
///         serves that tree one level at a time. It is therefore <b>inherently incomplete</b>: a topic
///         that never publishes during the window is invisible. That is a property of MQTT, not a
///         defect — the picker always keeps manual entry as the escape hatch.
///     </para>
///     <para>
///         <b>The load-bearing safety property is that browsing publishes nothing.</b> A picker opened
///         by an operator runs against a <i>live production broker</i>; if browse published anything,
///         opening an address picker could inject messages into a running plant. Every outbound
///         message must route through the single <see cref="PublishAsync"/> seam, which counts its
///         calls into <see cref="PublishCountForTest"/> so the guarantee is assertable.
///         <see cref="RequestRebirthAsync"/> — an explicit operator action, never a browse step — is
///         the <b>one and only</b> caller of that seam. <see cref="RootAsync"/>,
///         <see cref="ExpandAsync"/>, <see cref="AttributesAsync"/>, <see cref="Observe"/> and
///         <see cref="DisposeAsync"/> must never reach it, in either mode.
///     </para>
///     <para>
///         <b>Sparkplug mode serves a different tree, from the same passive window.</b> Instead of raw
///         <c>spBv1.0/…</c> topic segments — which look like a metric tree but are not one — the
///         session decodes observed NBIRTH/DBIRTH certificates into
///         Group → EdgeNode → [Device] → Metric, because a birth is the only Sparkplug message that
///         is self-describing (a DATA metric carries an alias and no name at all). The same node cap,
///         the same identifier bounds and the same "publishes nothing" guarantee apply.
///     </para>
///     <para>
///         Observations arrive on MQTTnet's dispatcher thread while the picker calls
///         <see cref="ExpandAsync"/> from the UI, so the tree is built from concurrent collections and
///         every mutation is lock-free.
///     </para>
/// </summary>
internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
{
    /// <summary>Default cap on recorded nodes; a busy broker's <c>#</c> firehose is unbounded.</summary>
    internal const int DefaultNodeCap = 50_000;

    /// <summary>Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet.</summary>
    internal const int PayloadSnippetMaxChars = 64;

    /// <summary>
    ///     Hard cap on the payload bytes fed to the UTF-8 decode. A <c>#</c> subscription receives
    ///     whatever the broker has — multi-MB retained blobs and JSON dumps included — and the node
    ///     cap does not help: a large payload on an <i>already-known</i> topic creates no node, it
    ///     just updates one, so an uncapped decode is unbounded work on MQTTnet's pump thread on
    ///     every single message, forever. 512 is <see cref="PayloadSnippetMaxChars"/> × 4 (UTF-8's
    ///     worst-case bytes per char, so the snippet can always be filled) doubled, leaving headroom
    ///     for leading whitespace ahead of a scalar.
    /// </summary>
    internal const int PayloadInspectMaxBytes = 512;

    /// <summary>
    ///     Cap on a whole topic name. MQTT permits ~65 KB, and the node cap bounds the node
    ///     <i>count</i> only — 50 000 nodes near that ceiling is a multi-GB tree. Topics past this
    ///     are rejected outright rather than truncated: a truncated topic is a <i>different</i>
    ///     topic, and the picker would commit it as a tag address.
    /// </summary>
    internal const int MaxTopicChars = 1024;

    /// <summary>Cap on one topic segment (one tree node's label + display name).</summary>
    internal const int MaxSegmentChars = 256;

    /// <summary>Sentinel node id of the "results truncated" marker, mirroring <c>CapturedTreeBrowseSession</c>.</summary>
    internal const string TruncatedNodeId = "__truncated__";

    /// <summary>Sentinel node id of the "oversized topics rejected" marker.</summary>
    internal const string OversizedNodeId = "__oversized__";

    /// <summary>
    ///     Separates a metric name from its owning group/edge-node/device path in a Sparkplug metric's
    ///     node id. A group/edge-node/device id is a single MQTT topic segment and therefore cannot
    ///     contain <c>/</c>, but a <b>metric name</b> routinely does (<c>Node Control/Rebirth</c>,
    ///     <c>Properties/Serial Number</c>) — so slash-joining would make a node-level metric
    ///     <c>Plant1/EdgeA/Temperature</c> indistinguishable from a device folder of the same shape,
    ///     and the picker would commit the wrong address. Nothing re-parses this: it exists so the ids
    ///     the tree hands out are unambiguous, while every rebirth target is read from the tuple stored
    ///     on the node itself.
    /// </summary>
    internal const string MetricSeparator = "::";

    /// <summary>
    ///     Ceiling on a Sparkplug birth body this session will decode. The plain path bounds its work
    ///     at <see cref="PayloadInspectMaxBytes"/>, but a birth must be parsed <i>whole</i> to yield its
    ///     metric list, so the bound has to be on the message instead. Mirrors
    ///     <c>MqttDriverOptions.MaxPayloadBytes</c>' 1 MiB default: the protobuf parse runs on MQTTnet's
    ///     shared dispatcher thread, so an unbounded one is unbounded work per message, forever.
    /// </summary>
    internal const int MaxBirthPayloadBytes = 1024 * 1024;

    /// <summary>
    ///     Cap on how many edge nodes one group-scoped <see cref="RequestRebirthAsync"/> will address.
    ///     A rebirth makes an edge node republish its entire metric set, so fanning one operator click
    ///     across a 200-node group is a self-inflicted broker storm. Past this the call is refused
    ///     <b>whole</b> — a partial fan-out would leave the operator unable to say which half of the
    ///     plant was asked to rebirth.
    /// </summary>
    internal const int MaxGroupRebirthNodes = 32;

    /// <summary>Bound on the recorded published-topic trail (a test/diagnostic aid, not a log).</summary>
    private const int PublishTrailMax = 64;

    /// <summary>
    ///     Throw-on-invalid so genuinely binary payloads are reported as such rather than silently
    ///     rendered as a field of replacement characters. Cached: allocating an encoding per message
    ///     would be per-message garbage on the dispatcher thread.
    /// </summary>
    private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true);

    /// <summary>Bound on the courtesy DISCONNECT issued at dispose — teardown never hangs on a dead broker.</summary>
    private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(5);

    private readonly IMqttClient? _client;
    private readonly ILogger? _logger;
    private readonly int _nodeCap;

    /// <summary>Flat path → node index, so <see cref="ExpandAsync"/> is O(1) rather than a tree walk.</summary>
    private readonly ConcurrentDictionary<string, TopicNode> _byPath = new(StringComparer.Ordinal);

    /// <summary>First-segment nodes, keyed by segment.</summary>
    private readonly ConcurrentDictionary<string, TopicNode> _roots = new(StringComparer.Ordinal);

    /// <summary>
    ///     The adapter that lets <see cref="RebirthRequester"/> reach this session's client
    ///     <i>through</i> <see cref="PublishAsync"/>. A private nested type rather than an interface on
    ///     the session itself, so <see cref="PublishAsync"/> stays private and the counted chokepoint
    ///     cannot be bypassed by a future caller holding the session.
    /// </summary>
    private readonly RebirthTransport _rebirthTransport;

    /// <summary>Topics this session has published, newest last; bounded by <see cref="PublishTrailMax"/>.</summary>
    private readonly ConcurrentQueue<string> _publishedTopics = new();

    private int _disposed;
    private long _lastUsedTicksUtc = DateTime.UtcNow.Ticks;
    private int _nodeCount;
    private int _oversized;
    private int _publishCount;
    private int _truncated;

    /// <summary>
    ///     Test seam: an unconnected session whose tree is fed through
    ///     <see cref="ObserveTopicForTest"/>. The accumulating tree is the whole of the P1 logic, so it
    ///     is exercised without a live broker.
    /// </summary>
    /// <param name="mode">The ingest shape the session was opened for.</param>
    /// <param name="nodeCap">Cap on recorded nodes; defaults to <see cref="DefaultNodeCap"/>.</param>
    internal MqttBrowseSession(MqttMode mode, int nodeCap = DefaultNodeCap)
        : this(mode, client: null, logger: null, nodeCap) { }

    /// <summary>Production constructor: takes ownership of an already-connected client.</summary>
    /// <param name="mode">The ingest shape the session was opened for.</param>
    /// <param name="client">The connected MQTT client; disconnected and disposed by <see cref="DisposeAsync"/>.</param>
    /// <param name="logger">Optional logger; never receives credentials or payload bodies.</param>
    /// <param name="nodeCap">Cap on recorded nodes; defaults to <see cref="DefaultNodeCap"/>.</param>
    internal MqttBrowseSession(MqttMode mode, IMqttClient? client, ILogger? logger, int nodeCap = DefaultNodeCap)
    {
        Mode = mode;
        _client = client;
        _logger = logger;
        _nodeCap = nodeCap;
        _rebirthTransport = new RebirthTransport(this);
    }

    /// <summary>
    ///     Number of messages this session has published — the assertion that browsing a live plant
    ///     broker is strictly read-only. <b>Always 0 unless <see cref="RequestRebirthAsync"/> was
    ///     explicitly invoked</b>: no browse path in either mode reaches the publish seam.
    /// </summary>
    internal int PublishCountForTest => Volatile.Read(ref _publishCount);

    /// <summary>The topics this session published, in order — the discriminating view of the above.</summary>
    internal IReadOnlyList<string> PublishedTopicsForTest => [.. _publishedTopics];

    /// <summary>True once the node cap stopped recording; surfaced as a marker node from <see cref="RootAsync"/>.</summary>
    internal bool Truncated => Volatile.Read(ref _truncated) != 0;

    /// <summary>
    ///     True once a topic was rejected for breaching <see cref="MaxTopicChars"/> /
    ///     <see cref="MaxSegmentChars"/>; surfaced as its own marker node from
    ///     <see cref="RootAsync"/>. Kept distinct from <see cref="Truncated"/> because the operator's
    ///     remedy differs — narrow the prefix, versus this broker publishes absurd topic names.
    /// </summary>
    internal bool OversizedRejected => Volatile.Read(ref _oversized) != 0;

    private bool Disposed => Volatile.Read(ref _disposed) != 0;

    /// <summary>
    ///     The ingest shape this session was opened for. P1 serves <see cref="MqttMode.Plain"/> only;
    ///     the P2 Sparkplug tasks branch the tree shape (Group → EdgeNode → Device → Metric) on it.
    /// </summary>
    internal MqttMode Mode { get; }

    /// <inheritdoc />
    public Guid Token { get; } = Guid.NewGuid();

    /// <inheritdoc />
    public DateTime LastUsedUtc => new(Interlocked.Read(ref _lastUsedTicksUtc), DateTimeKind.Utc);

    /// <inheritdoc />
    public Task<IReadOnlyList<BrowseNode>> RootAsync(CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(Disposed, this);
        var nodes = Project(_roots);
        if (Truncated)
        {
            nodes.Add(new BrowseNode(
                TruncatedNodeId,
                "⚠ Observation truncated — narrow the topic prefix or use manual entry",
                BrowseNodeKind.Folder,
                HasChildrenHint: false));
        }

        if (OversizedRejected)
        {
            nodes.Add(new BrowseNode(
                OversizedNodeId,
                Mode == MqttMode.SparkplugB
                    ? "⚠ Some metrics were unusable (oversized or malformed names) — use manual entry for those"
                    : "⚠ Some topics were too long to show — use manual entry for those",
                BrowseNodeKind.Folder,
                HasChildrenHint: false));
        }

        Touch();
        return Task.FromResult<IReadOnlyList<BrowseNode>>(nodes);
    }

    /// <inheritdoc />
    /// <remarks>
    ///     <paramref name="nodeId"/> is the full slash-joined topic path of the node
    ///     (e.g. <c>factory/line3/oven</c>) — the same value carried on the <see cref="BrowseNode"/>
    ///     the picker was handed. An unknown path returns empty rather than throwing: the tree grows
    ///     while the picker browses it, so a stale id is an ordinary race, not an error.
    /// </remarks>
    public Task<IReadOnlyList<BrowseNode>> ExpandAsync(string nodeId, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(Disposed, this);
        var children = _byPath.TryGetValue(nodeId ?? "", out var node)
            ? Project(node.Children)
            : [];
        Touch();
        return Task.FromResult<IReadOnlyList<BrowseNode>>(children);
    }

    /// <inheritdoc />
    /// <remarks>
    ///     <para>
    ///         <b>Plain mode.</b> A topic has no attribute model — MQTT carries an opaque payload. The
    ///         single synthetic attribute returned here exists to help the operator confirm they picked
    ///         the right topic: <see cref="AttributeInfo.DriverDataType"/> carries the type
    ///         <i>inferred</i> from the last observed payload (the authored tag's own <c>dataType</c>
    ///         remains the authority — inference is the brittle fallback, per design §4), and
    ///         <see cref="AttributeInfo.SecurityClass"/> carries that payload's snippet. The
    ///         <c>SecurityClass</c> slot is reused deliberately: it is the picker's free-text descriptor
    ///         line (rendered as <c>{DriverDataType} · {SecurityClass}</c>) and MQTT has no security
    ///         classes of its own. Returns empty for a synthesised path segment nothing was ever
    ///         published to.
    ///     </para>
    ///     <para>
    ///         <b>Sparkplug mode is purely birth-derived — no payload is ever sniffed.</b> A Sparkplug
    ///         metric <i>declares</i> its own type in the birth certificate, so there is nothing to
    ///         infer; and the plain path's UTF-8 inference / snippet machinery is not merely redundant
    ///         here but actively wrong, because a Sparkplug body is protobuf and would be reported as
    ///         <c>(N bytes, binary)</c> for every metric in the plant. This session therefore retains no
    ///         payload bytes at all in Sparkplug mode: the attribute carries the metric name, the
    ///         <see cref="DriverDataType"/> the birth declared (element type for an array variant, with
    ///         <see cref="AttributeInfo.IsArray"/> carrying the array bit), and a descriptor built from
    ///         the birth. A datatype this driver does not support in v1 is reported as such and
    ///         <b>never</b> coerced into a guessed <see cref="DriverDataType"/>.
    ///     </para>
    /// </remarks>
    public Task<IReadOnlyList<AttributeInfo>> AttributesAsync(string nodeId, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(Disposed, this);

        IReadOnlyList<AttributeInfo> attrs = [];
        if (_byPath.TryGetValue(nodeId ?? "", out var node))
        {
            if (node.Metric is { } metric)
            {
                attrs = [DescribeMetric(metric)];
            }
            else if (node.Observed is { } observed)
            {
                attrs = [new AttributeInfo(node.Segment, observed.DataType.ToString(), IsArray: false, observed.Snippet)];
            }
        }

        Touch();
        return Task.FromResult(attrs);
    }

    /// <summary>
    ///     Projects one observed Sparkplug birth metric into the picker's attribute row.
    /// </summary>
    /// <param name="metric">The metric as its birth certificate declared it.</param>
    /// <returns>The attribute row.</returns>
    /// <remarks>
    ///     <see cref="SparkplugDataTypeExtensions.ToDriverDataType"/> returns <see langword="null"/> for
    ///     the five v1-unsupported datatypes (<c>Unknown</c>, <c>DataSet</c>, <c>Template</c>,
    ///     <c>PropertySet</c>, <c>PropertySetList</c>), and that null is passed through as the Sparkplug
    ///     type's own name rather than defaulted. None of those five names parses as a
    ///     <see cref="DriverDataType"/>, so the AdminUI's browse-commit mapper falls back to its
    ///     declared default instead of silently binding a metric this driver cannot serve — and the
    ///     descriptor line says so in words.
    /// </remarks>
    private static AttributeInfo DescribeMetric(SparkplugMetricInfo metric)
    {
        var mapped = metric.DataType?.ToDriverDataType();
        var sparkplugName = metric.DataType?.ToString() ?? nameof(SparkplugDataType.Unknown);

        var descriptor = new StringBuilder(metric.BirthType.ToString()).Append(" · ").Append(sparkplugName);
        if (metric.Alias is { } alias) descriptor.Append(" · alias ").Append(alias);
        if (mapped is null) descriptor.Append(" · unsupported in v1 — not bindable");

        return new AttributeInfo(
            metric.Name,
            mapped?.ToString() ?? sparkplugName,
            metric.DataType?.IsSparkplugArray() ?? false,
            descriptor.ToString());
    }

    /// <summary>
    ///     Best-effort teardown. Safe on a session that was never opened (test seam, or an
    ///     <c>OpenAsync</c> that failed between constructing the client and connecting it) and
    ///     idempotent under the reaper racing the picker's own close.
    /// </summary>
    /// <returns>A task that represents the asynchronous teardown.</returns>
    public async ValueTask DisposeAsync()
    {
        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;

        if (_client is { } client)
        {
            try
            {
                using var cts = new CancellationTokenSource(DisconnectTimeout);
                await client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                // The broker may already be gone, or never have answered the CONNECT at all.
                _logger?.LogDebug(ex, "MQTT browse session disconnect failed; client abandoned.");
            }

            try { client.Dispose(); }
            catch (Exception ex) { _logger?.LogDebug(ex, "MQTT browse client dispose failed."); }
        }

        _byPath.Clear();
        _roots.Clear();
    }

    /// <summary>
    ///     Records one observed message into the accumulating tree, in whichever shape
    ///     <see cref="Mode"/> selects. Called from MQTTnet's dispatcher thread, so it does no I/O,
    ///     takes no lock, and never throws — a throwing handler would stall the client's pump. It also
    ///     never publishes: this is an observation seam, and the browser's read-only guarantee depends
    ///     on nothing here reaching <see cref="PublishAsync"/>.
    /// </summary>
    /// <param name="topic">The topic the message arrived on.</param>
    /// <param name="payload">The message body.</param>
    internal void Observe(string topic, ReadOnlySpan<byte> payload)
    {
        if (Disposed || string.IsNullOrWhiteSpace(topic)) return;

        if (Mode == MqttMode.SparkplugB)
        {
            ObserveSparkplugBirth(topic, payload);
            return;
        }

        ObservePlainTopic(topic, payload);
    }

    /// <summary>
    ///     Plain mode: records the topic as a path in the segment tree, decorated with the payload's
    ///     inferred type and display snippet.
    /// </summary>
    /// <param name="topic">The topic the message arrived on.</param>
    /// <param name="payload">The message body; used only for type inference and the display snippet.</param>
    private void ObservePlainTopic(string topic, ReadOnlySpan<byte> payload)
    {
        if (topic.Length > MaxTopicChars)
        {
            Volatile.Write(ref _oversized, 1);
            return;
        }

        var segments = topic.Split('/');

        // Checked as a pre-pass so a rejected topic leaves no half-built path behind.
        foreach (var segment in segments)
        {
            if (segment.Length > MaxSegmentChars)
            {
                Volatile.Write(ref _oversized, 1);
                return;
            }
        }

        var level = _roots;
        TopicNode? node = null;
        var path = string.Empty;

        foreach (var segment in segments)
        {
            path = path.Length == 0 ? segment : path + "/" + segment;

            node = GetOrCreate(level, path, segment, kind: null);
            if (node is null) return;   // node cap tripped; already-recorded topics keep serving

            level = node.Children;
        }

        node?.Record(InferPayload(payload));
    }

    /// <summary>
    ///     Sparkplug mode: decodes an observed <b>birth certificate</b> into the
    ///     Group → EdgeNode → [Device] → Metric tree.
    /// </summary>
    /// <param name="topic">The topic the message arrived on.</param>
    /// <param name="payload">The Sparkplug-B protobuf body.</param>
    /// <remarks>
    ///     <para>
    ///         <b>Births only.</b> NBIRTH/DBIRTH are the only Sparkplug messages that name their
    ///         metrics; a DATA metric carries an alias and no name at all (that is the whole point of
    ///         the alias table), so a tree built from DATA would be a tree of unnamed nodes. NDEATH,
    ///         DDEATH, NCMD, DCMD and STATE carry no metric model worth browsing, and everything that
    ///         is not a Sparkplug topic at all is ignored outright rather than falling back to the
    ///         plain segment tree — a raw <c>spBv1.0/…</c> tree looks like a metric tree but is not one.
    ///     </para>
    ///     <para>
    ///         <b>Node-level metrics hang directly under their edge node</b>, not under a synthesised
    ///         "(node)" device folder. The authored address tuple is
    ///         <c>(group, edgeNode, device?, metric)</c> with <c>device</c> genuinely absent for these,
    ///         so an invented folder segment would be indistinguishable from a real device of that
    ///         name — and the picker would commit an address that does not exist. Devices render as
    ///         folders and metrics as leaves, so the two stay distinguishable side by side.
    ///     </para>
    /// </remarks>
    private void ObserveSparkplugBirth(string topic, ReadOnlySpan<byte> payload)
    {
        if (!SparkplugTopic.TryParse(topic, out var parsed)) return;
        if (parsed.Type is not (SparkplugMessageType.NBIRTH or SparkplugMessageType.DBIRTH)) return;

        // The bound has to be on the whole message: unlike a plain payload, a birth must be parsed
        // entire to yield its metric list, and the parse runs on the shared dispatcher thread.
        if (payload.Length > MaxBirthPayloadBytes)
        {
            Volatile.Write(ref _oversized, 1);
            return;
        }

        // Never throws, for any input — see the SparkplugCodec type remarks.
        if (!SparkplugCodec.TryDecode(payload, out var decoded)) return;

        var groupId = parsed.GroupId!;
        var edgeNodeId = parsed.EdgeNodeId!;
        var deviceId = parsed.DeviceId;

        if (!IsUsableLabel(groupId) || !IsUsableLabel(edgeNodeId) || (deviceId is not null && !IsUsableLabel(deviceId)))
        {
            Volatile.Write(ref _oversized, 1);
            return;
        }

        var scope = new SparkplugScope(groupId, edgeNodeId);

        var group = GetOrCreate(_roots, groupId, groupId, BrowseNodeKind.Folder, new SparkplugScope(groupId, null));
        if (group is null) return;

        var edgePath = groupId + "/" + edgeNodeId;
        var edge = GetOrCreate(group.Children, edgePath, edgeNodeId, BrowseNodeKind.Folder, scope);
        if (edge is null) return;

        var owner = edge;
        var ownerPath = edgePath;
        if (deviceId is not null)
        {
            ownerPath = edgePath + "/" + deviceId;
            owner = GetOrCreate(edge.Children, ownerPath, deviceId, BrowseNodeKind.Folder, scope);
            if (owner is null) return;
        }

        foreach (var metric in decoded.Metrics)
        {
            // A birth is REQUIRED to name every metric it carries; one that does not is not something
            // a picker can commit an address for.
            var name = metric.Name;
            if (name is null || !IsUsableLabel(name))
            {
                if (name is not null) Volatile.Write(ref _oversized, 1);
                continue;
            }

            var metricPath = ownerPath + MetricSeparator + name;
            if (metricPath.Length > MaxTopicChars)
            {
                Volatile.Write(ref _oversized, 1);
                continue;
            }

            var node = GetOrCreate(owner.Children, metricPath, name, BrowseNodeKind.Leaf, scope);
            if (node is null) return;   // node cap tripped

            node.RecordMetric(new SparkplugMetricInfo(name, metric.DataType, parsed.Type, metric.Alias));
        }
    }

    /// <summary>
    ///     Finds or creates one tree node under <paramref name="level"/>, honouring the node cap.
    /// </summary>
    /// <param name="level">The child map to place the node in.</param>
    /// <param name="path">The node's full path — its browse node id, and its <see cref="_byPath"/> key.</param>
    /// <param name="segment">The node's own label.</param>
    /// <param name="kind">
    ///     An explicit browse kind, or <see langword="null"/> to let <see cref="Project"/> infer it from
    ///     whether the node has children. Sparkplug sets it explicitly: a device whose birth carried no
    ///     metrics is childless but is still a folder, and must not masquerade as a committable metric.
    /// </param>
    /// <param name="scope">The Sparkplug scope to stamp on a newly created node, if any.</param>
    /// <returns>The node, or <see langword="null"/> when the node cap stopped recording.</returns>
    private TopicNode? GetOrCreate(
        ConcurrentDictionary<string, TopicNode> level,
        string path,
        string segment,
        BrowseNodeKind? kind,
        SparkplugScope? scope = null)
    {
        // Keyed by full path, not by label. In plain mode the two are equivalent (a path is its
        // parent's path plus its label), but in Sparkplug mode a device and an edge-node-level metric
        // are siblings that can share a label — `Plant1/EdgeA/Temp` and `Plant1/EdgeA::Temp` are
        // different nodes, and keying by label would silently drop whichever arrived second.
        if (level.TryGetValue(path, out var existing)) return existing;

        if (Volatile.Read(ref _nodeCount) >= _nodeCap)
        {
            Volatile.Write(ref _truncated, 1);
            return null;
        }

        var created = new TopicNode(path, segment, kind, scope);
        var child = level.GetOrAdd(path, created);
        if (ReferenceEquals(child, created))
        {
            Interlocked.Increment(ref _nodeCount);
            _byPath[path] = child;
        }

        return child;
    }

    /// <summary>
    ///     Whether an identifier off the wire is usable as a tree label. Length-bounded like a topic
    ///     segment, and control-character-free: a metric name is arbitrary UTF-8 chosen by the
    ///     publisher, and an embedded newline or ANSI escape would break the picker's single-line rows.
    ///     Unlike the payload snippet — which is scrubbed — a name is <b>rejected</b> rather than
    ///     cleaned, because a scrubbed name is a different name and the picker would commit it as the
    ///     tag's binding key.
    /// </summary>
    /// <param name="label">The candidate label.</param>
    /// <returns><see langword="true"/> when the label may be recorded.</returns>
    private static bool IsUsableLabel(string label)
    {
        if (label.Length == 0 || label.Length > MaxSegmentChars) return false;
        foreach (var c in label)
        {
            if (char.IsControl(c)) return false;
        }

        return true;
    }

    /// <summary>Test seam over <see cref="Observe"/> that takes the payload as text.</summary>
    /// <param name="topic">The observed topic.</param>
    /// <param name="payload">The message body as text, or <see langword="null"/> for an empty payload.</param>
    internal void ObserveTopicForTest(string topic, string? payload = null) =>
        Observe(topic, payload is null ? ReadOnlySpan<byte>.Empty : Encoding.UTF8.GetBytes(payload));

    /// <summary>Test seam over <see cref="Observe"/> that takes raw wire bytes.</summary>
    /// <param name="topic">The observed topic.</param>
    /// <param name="payload">The raw message body.</param>
    internal void ObserveRawForTest(string topic, byte[] payload) => Observe(topic, payload);

    /// <summary>
    ///     Test seam that feeds one metric's birth certificate through the <b>real</b> observe path —
    ///     it encodes an actual Sparkplug-B payload and hands it to <see cref="Observe"/>, so the topic
    ///     parse, the protobuf decode and the tree build are all exercised rather than mocked around.
    /// </summary>
    /// <param name="groupId">The Sparkplug group id.</param>
    /// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
    /// <param name="device">The device id, or <see langword="null"/> for an edge-node-level metric (NBIRTH).</param>
    /// <param name="metricName">The metric name the birth declares.</param>
    /// <param name="dataType">The datatype the birth declares.</param>
    /// <param name="alias">The metric alias the birth declares, if any.</param>
    internal void ObserveBirthForTest(
        string groupId,
        string edgeNodeId,
        string? device,
        string metricName,
        SparkplugDataType dataType,
        ulong? alias = null)
    {
        var metric = new Payload.Types.Metric { Name = metricName, Datatype = (uint)dataType };
        if (alias is { } a) metric.Alias = a;
        ObserveBirth(groupId, edgeNodeId, device, metric);
    }

    /// <summary>Test seam feeding a birth certificate that declares no metrics at all.</summary>
    /// <param name="groupId">The Sparkplug group id.</param>
    /// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
    /// <param name="device">The device id, or <see langword="null"/> for an NBIRTH.</param>
    internal void ObserveEmptyBirthForTest(string groupId, string edgeNodeId, string? device) =>
        ObserveBirth(groupId, edgeNodeId, device, metric: null);

    /// <summary>Encodes and observes one birth certificate.</summary>
    /// <param name="groupId">The Sparkplug group id.</param>
    /// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
    /// <param name="device">The device id, or <see langword="null"/> for an NBIRTH.</param>
    /// <param name="metric">The single metric the birth carries, or <see langword="null"/> for none.</param>
    private void ObserveBirth(string groupId, string edgeNodeId, string? device, Payload.Types.Metric? metric)
    {
        var type = device is null ? SparkplugMessageType.NBIRTH : SparkplugMessageType.DBIRTH;
        var topic = SparkplugTopic.Format(groupId, type, edgeNodeId, device);

        var payload = new Payload { Seq = 0 };
        if (metric is not null) payload.Metrics.Add(metric);

        Observe(topic, payload.ToByteArray());
    }

    /// <inheritdoc />
    /// <remarks>
    ///     <para>
    ///         <b>The one browse-session member that publishes, and it is never on a browse path.</b>
    ///         Only an explicit operator click reaches it — <see cref="OpenAsync"/>-equivalent setup,
    ///         <see cref="RootAsync"/>, <see cref="ExpandAsync"/>, <see cref="AttributesAsync"/>,
    ///         <see cref="Observe"/> and <see cref="DisposeAsync"/> must never call it, and
    ///         <see cref="PublishCountForTest"/> is what makes that assertable rather than aspirational.
    ///         Authorization is the caller's job (<c>BrowserSessionService</c> enforces
    ///         <c>DriverOperator</c>): this type holds no user identity.
    ///     </para>
    ///     <para>
    ///         <b>Scope resolution never re-parses an id string when it does not have to.</b> A scope
    ///         that names a node in this session's own tree resolves through the
    ///         <c>(group, edgeNode)</c> tuple stamped on that node — so selecting a device or a metric
    ///         addresses its <i>owning edge node</i>, which is the only thing an NCMD can address
    ///         (Sparkplug has no device-scoped rebirth). A two-segment <c>{group}/{edgeNode}</c> scope
    ///         is honoured even when the window has never observed that node: a node that has not
    ///         birthed is the <i>prime</i> rebirth target, and refusing it would defeat the feature.
    ///     </para>
    ///     <para>
    ///         <b>Group scope is bounded and all-or-nothing</b> — see
    ///         <see cref="MaxGroupRebirthNodes"/>.
    ///     </para>
    /// </remarks>
    public async Task<int> RequestRebirthAsync(string scope, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(Disposed, this);

        if (Mode != MqttMode.SparkplugB)
        {
            throw new NotSupportedException(
                "A rebirth request is a Sparkplug B action; a plain-MQTT browse window publishes nothing, ever.");
        }

        ArgumentException.ThrowIfNullOrWhiteSpace(scope);

        var targets = ResolveRebirthTargets(scope);

        foreach (var (groupId, edgeNodeId) in targets)
        {
            // QoS 0, retain false, bounded deadline — all owned by RebirthRequester, which reaches the
            // client only through this session's counted PublishAsync chokepoint.
            await RebirthRequester
                .RequestAsync(_rebirthTransport, groupId, edgeNodeId, cancellationToken)
                .ConfigureAwait(false);
        }

        Touch();
        return targets.Count;
    }

    /// <summary>
    ///     Resolves a rebirth scope to the edge nodes an NCMD should be addressed to. Pure — it
    ///     publishes nothing, so an over-wide or unresolvable scope is refused before any message
    ///     leaves.
    /// </summary>
    /// <param name="scope">The scope, as documented on <see cref="RequestRebirthAsync"/>.</param>
    /// <returns>The distinct <c>(group, edgeNode)</c> targets, ordered for determinism.</returns>
    /// <exception cref="ArgumentException">The scope is not resolvable to any edge node.</exception>
    /// <exception cref="InvalidOperationException">
    ///     A group scope named no observed edge node, or more than <see cref="MaxGroupRebirthNodes"/>.
    /// </exception>
    private IReadOnlyList<(string GroupId, string EdgeNodeId)> ResolveRebirthTargets(string scope)
    {
        // 1. A node from this session's own tree — the normal path. The target comes off the stored
        //    tuple, never off the id string.
        if (_byPath.TryGetValue(scope, out var node) && node.Scope is { } nodeScope)
        {
            if (nodeScope.EdgeNodeId is { } edgeNodeId) return [(nodeScope.GroupId, edgeNodeId)];
            return EnumerateGroup(nodeScope.GroupId, node);
        }

        var segments = scope.Split('/');

        // 2. A group that is a root of this tree (identical to case 1 for an observed group, and
        //    reached only when the group node itself was evicted by the node cap).
        if (segments.Length == 1 && _roots.TryGetValue(segments[0], out var group))
        {
            return EnumerateGroup(segments[0], group);
        }

        // 3. A bare {group}/{edgeNode} the window has never seen — the prime rebirth target.
        if (segments.Length == 2 && IsUsableLabel(segments[0]) && IsUsableLabel(segments[1]))
        {
            return [(segments[0], segments[1])];
        }

        throw new ArgumentException(
            $"'{scope}' does not name a Sparkplug group or edge node in this browse session; "
            + "select a node in the tree, or address one as '{group}/{edgeNode}'.",
            nameof(scope));
    }

    /// <summary>Enumerates the observed edge nodes of one group, bounded by <see cref="MaxGroupRebirthNodes"/>.</summary>
    /// <param name="groupId">The group id.</param>
    /// <param name="group">The group's tree node.</param>
    /// <returns>The group's edge-node targets, ordered by edge-node id.</returns>
    private static IReadOnlyList<(string GroupId, string EdgeNodeId)> EnumerateGroup(string groupId, TopicNode group)
    {
        var edgeNodes = group.Children.Values
            .Select(n => n.Scope?.EdgeNodeId)
            .Where(id => id is not null)
            .Select(id => id!)
            .Distinct(StringComparer.Ordinal)
            .OrderBy(id => id, StringComparer.Ordinal)
            .ToList();

        if (edgeNodes.Count == 0)
        {
            throw new InvalidOperationException(
                $"No edge node has been observed in Sparkplug group '{groupId}', so there is nothing to "
                + "address; select an edge node, or address one as '{group}/{edgeNode}'.");
        }

        if (edgeNodes.Count > MaxGroupRebirthNodes)
        {
            throw new InvalidOperationException(
                $"Sparkplug group '{groupId}' has {edgeNodes.Count} observed edge nodes, more than the "
                + $"{MaxGroupRebirthNodes} one request will fan out to; select a single edge node instead. "
                + "Nothing was published.");
        }

        return [.. edgeNodes.Select(id => (groupId, id))];
    }

    /// <summary>
    ///     <b>The one and only outbound-message seam.</b> Every browse path is strictly read-only, and
    ///     <see cref="PublishCountForTest"/> is the assertion of that;
    ///     <see cref="RequestRebirthAsync"/> — an explicit operator action — is the single sanctioned
    ///     caller. Anything that ever needs to publish <b>must route through here</b> rather than
    ///     touching <see cref="_client"/> directly, or the guarantee stops being checkable.
    /// </summary>
    /// <param name="message">The message to publish.</param>
    /// <param name="cancellationToken">Cancellation token for the publish.</param>
    /// <returns>A task that represents the asynchronous publish.</returns>
    private async Task PublishAsync(MqttApplicationMessage message, CancellationToken cancellationToken)
    {
        Interlocked.Increment(ref _publishCount);

        _publishedTopics.Enqueue(message.Topic);
        while (_publishedTopics.Count > PublishTrailMax) _publishedTopics.TryDequeue(out _);

        if (_client is { } client)
        {
            await client.PublishAsync(message, cancellationToken).ConfigureAwait(false);
        }
    }

    /// <summary>
    ///     Best-effort type inference from a payload body. Explicit authoring stays the authority
    ///     (design §4) — this only decorates the picker. Non-UTF-8 bodies are reported as opaque.
    ///     <para>
    ///         Runs on MQTTnet's dispatcher thread, so the work is bounded before any decode: at most
    ///         <see cref="PayloadInspectMaxBytes"/> bytes are ever examined, however large the message.
    ///     </para>
    /// </summary>
    /// <param name="payload">The raw message body.</param>
    /// <returns>The inferred type and a bounded display snippet.</returns>
    private static ObservedValue InferPayload(ReadOnlySpan<byte> payload)
    {
        if (payload.IsEmpty) return new ObservedValue(DriverDataType.String, "(empty)");

        var totalBytes = payload.Length;
        var clipped = totalBytes > PayloadInspectMaxBytes;
        // A blind slice can cut a multi-byte sequence in half, which the strict decoder would report
        // as "binary" — mislabelling a perfectly good UTF-8 payload. Back the window off to the last
        // whole sequence instead.
        var window = clipped ? TrimToUtf8Boundary(payload[..PayloadInspectMaxBytes]) : payload;

        string text;
        try
        {
            text = Utf8Strict.GetString(window);
        }
        catch (DecoderFallbackException)
        {
            return new ObservedValue(DriverDataType.String, $"({totalBytes} bytes, binary)");
        }

        var trimmed = text.Trim();

        // Ordered narrowest-first. "1" infers Int64 rather than Boolean: an integer reading is the
        // commoner meaning and, unlike the reverse, does not silently collapse a range to two states.
        // A clipped payload is never a scalar, and inferring from a partial view could read a
        // whitespace-padded prefix as a number it is not — so it is String by construction.
        var dataType = clipped
            ? DriverDataType.String
            : trimmed switch
            {
                _ when bool.TryParse(trimmed, out _) => DriverDataType.Boolean,
                _ when long.TryParse(trimmed, NumberStyles.Integer, CultureInfo.InvariantCulture, out _)
                    => DriverDataType.Int64,
                _ when double.TryParse(trimmed, NumberStyles.Float, CultureInfo.InvariantCulture, out _)
                    => DriverDataType.Float64,
                _ => DriverDataType.String,
            };

        return new ObservedValue(dataType, Sanitize(trimmed, clipped));
    }

    /// <summary>
    ///     Renders decoded payload text as a single-line, length-bounded snippet. Control characters
    ///     (newlines, tabs, ANSI escapes) become spaces — <see cref="string.Trim()"/> only strips them
    ///     at the ends, and an embedded one would break the picker's single-line attribute row.
    /// </summary>
    /// <param name="text">The trimmed decoded text.</param>
    /// <param name="clipped">Whether bytes beyond the inspection window were discarded.</param>
    /// <returns>The display snippet.</returns>
    private static string Sanitize(string text, bool clipped)
    {
        var overLength = text.Length > PayloadSnippetMaxChars;
        var span = overLength ? text.AsSpan(0, PayloadSnippetMaxChars) : text.AsSpan();

        // clipped, not just overLength: a huge payload whose first 512 bytes are mostly whitespace
        // yields a short snippet that would otherwise claim to be the whole message.
        return overLength || clipped ? Scrub(span) + "…" : Scrub(span);
    }

    /// <summary>Replaces control characters with spaces and trims the trailing edge.</summary>
    /// <param name="span">The characters to scrub.</param>
    /// <returns>The scrubbed text.</returns>
    private static string Scrub(ReadOnlySpan<char> span)
    {
        Span<char> buffer = span.Length <= 128 ? stackalloc char[span.Length] : new char[span.Length];
        for (var i = 0; i < span.Length; i++) buffer[i] = char.IsControl(span[i]) ? ' ' : span[i];
        return new string(buffer).TrimEnd();
    }

    /// <summary>
    ///     Shortens a byte window so it ends on a whole UTF-8 sequence. Walks back over continuation
    ///     bytes (<c>10xxxxxx</c>) to the lead byte and drops it when the sequence it starts runs past
    ///     the window. A stray continuation byte is left in place — that is genuine corruption, and the
    ///     strict decoder should say so.
    /// </summary>
    /// <param name="bytes">The blindly-sliced window.</param>
    /// <returns>The window, shortened to a sequence boundary.</returns>
    private static ReadOnlySpan<byte> TrimToUtf8Boundary(ReadOnlySpan<byte> bytes)
    {
        var i = bytes.Length - 1;
        var walked = 0;
        while (i >= 0 && (bytes[i] & 0xC0) == 0x80 && walked < 3)
        {
            i--;
            walked++;
        }

        if (i < 0) return ReadOnlySpan<byte>.Empty; // nothing but continuation bytes

        var needed = bytes[i] switch
        {
            < 0x80 => 1,   // ASCII
            >= 0xF0 => 4,
            >= 0xE0 => 3,
            >= 0xC0 => 2,
            _ => 1,        // stray continuation byte — let the decoder reject it
        };

        return bytes.Length - i >= needed ? bytes : bytes[..i];
    }

    /// <summary>Projects one level of the tree into browse nodes, ordered so the picker is stable.</summary>
    /// <param name="level">The child map to project.</param>
    /// <returns>The projected nodes, sorted by segment.</returns>
    private static List<BrowseNode> Project(ConcurrentDictionary<string, TopicNode> level) =>
        level.Values
            .OrderBy(n => n.Segment, StringComparer.Ordinal)
            .Select(n =>
            {
                // Plain mode infers: a childless node is always the terminal segment of a topic that
                // actually published, so it is the committable address, while a node that both
                // published and has children below it renders as a folder — MQTT allows that shape and
                // the picker must stay expandable. Sparkplug states the kind outright, because a
                // device whose birth carried no metrics is childless and must still not be offered as
                // a committable metric.
                var hasChildren = !n.Children.IsEmpty;
                var kind = n.Kind ?? (hasChildren ? BrowseNodeKind.Folder : BrowseNodeKind.Leaf);
                return new BrowseNode(
                    n.Path,
                    n.Segment,
                    kind,
                    hasChildren || kind == BrowseNodeKind.Folder);
            })
            .ToList();

    private void Touch() => Interlocked.Exchange(ref _lastUsedTicksUtc, DateTime.UtcNow.Ticks);

    /// <summary>The most recent payload observed on a topic, published as one atomic reference.</summary>
    /// <param name="DataType">The type inferred from that payload.</param>
    /// <param name="Snippet">A bounded, display-safe rendering of that payload.</param>
    private sealed record ObservedValue(DriverDataType DataType, string Snippet);

    /// <summary>
    ///     The Sparkplug identity a tree node belongs to. Carried explicitly on the node so a rebirth
    ///     target is <b>read</b>, never parsed back out of a node-id string — the same discipline the
    ///     v3 address space applies to <c>AddressSpaceRealm</c>.
    /// </summary>
    /// <param name="GroupId">The Sparkplug group id.</param>
    /// <param name="EdgeNodeId">
    ///     The owning edge-node id, or <see langword="null"/> on a group node (which addresses every
    ///     edge node beneath it).
    /// </param>
    private sealed record SparkplugScope(string GroupId, string? EdgeNodeId);

    /// <summary>One metric, exactly as a birth certificate declared it. No payload bytes are retained.</summary>
    /// <param name="Name">The metric name — the tag's stable binding key across rebirths.</param>
    /// <param name="DataType">
    ///     The declared Sparkplug datatype, or <see langword="null"/> when the birth omitted it (which a
    ///     conformant birth never does; kept nullable because the decoder preserves absence rather than
    ///     defaulting it).
    /// </param>
    /// <param name="BirthType">Which birth declared it — <c>NBIRTH</c> or <c>DBIRTH</c>.</param>
    /// <param name="Alias">The alias the birth assigned, if any. Display only: aliases are per-birth
    /// and may be reused for a different metric after a rebirth, so nothing binds by them.</param>
    private sealed record SparkplugMetricInfo(
        string Name,
        SparkplugDataType? DataType,
        SparkplugMessageType BirthType,
        ulong? Alias);

    /// <summary>One node of the observed tree — a topic segment in plain mode, a group / edge node /
    /// device / metric in Sparkplug mode.</summary>
    private sealed class TopicNode(string path, string segment, BrowseNodeKind? kind = null, SparkplugScope? scope = null)
    {
        /// <summary>Full path — the node id the picker round-trips.</summary>
        public string Path { get; } = path;

        /// <summary>This node's own label.</summary>
        public string Segment { get; } = segment;

        /// <summary>
        ///     An explicit browse kind, or <see langword="null"/> to infer it from the child count.
        ///     Set in Sparkplug mode only; see <see cref="Project"/>.
        /// </summary>
        public BrowseNodeKind? Kind { get; } = kind;

        /// <summary>The Sparkplug identity this node belongs to; <see langword="null"/> in plain mode.</summary>
        public SparkplugScope? Scope { get; } = scope;

        /// <summary>Child nodes, keyed by their full path (see <see cref="GetOrCreate"/>).</summary>
        public ConcurrentDictionary<string, TopicNode> Children { get; } = new(StringComparer.Ordinal);

        /// <summary>
        ///     The last payload observed on exactly this topic, or <see langword="null"/> when this is
        ///     a synthesised intermediate segment nothing published to. Written by the dispatcher
        ///     thread and read by the UI thread, so the whole record is swapped as one reference.
        ///     Plain mode only — Sparkplug retains no payload bytes.
        /// </summary>
        public ObservedValue? Observed => Volatile.Read(ref _observed);

        /// <summary>
        ///     The birth-declared metric this node is, or <see langword="null"/> when it is a folder
        ///     (or a plain-mode topic segment). Swapped as one reference for the same reason as
        ///     <see cref="Observed"/> — a rebirth republishes the whole set, redeclaring types.
        /// </summary>
        public SparkplugMetricInfo? Metric => Volatile.Read(ref _metric);

        private ObservedValue? _observed;
        private SparkplugMetricInfo? _metric;

        /// <summary>Publishes the latest observation for this topic.</summary>
        /// <param name="value">The inferred type + snippet to record.</param>
        public void Record(ObservedValue value) => Volatile.Write(ref _observed, value);

        /// <summary>Publishes the latest birth declaration for this metric.</summary>
        /// <param name="value">The metric as the most recent birth declared it.</param>
        public void RecordMetric(SparkplugMetricInfo value) => Volatile.Write(ref _metric, value);
    }

    /// <summary>
    ///     Adapts <see cref="RebirthRequester"/>'s publish seam onto this session's single counted
    ///     <see cref="PublishAsync"/> chokepoint. A private nested type on purpose: making the session
    ///     itself an <see cref="IMqttPublishTransport"/> would hand anyone holding it a general
    ///     publish-anything capability, and the read-only guarantee rests on there being exactly one
    ///     way out.
    /// </summary>
    /// <param name="owner">The session whose client the message is published on.</param>
    private sealed class RebirthTransport(MqttBrowseSession owner) : IMqttPublishTransport
    {
        /// <inheritdoc />
        public Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken)
        {
            var message = new MqttApplicationMessageBuilder()
                .WithTopic(topic)
                .WithPayload(payload)
                .WithQualityOfServiceLevel((MQTTnet.Protocol.MqttQualityOfServiceLevel)Math.Clamp(qos, 0, 2))
                .WithRetainFlag(retain)
                .Build();

            return owner.PublishAsync(message, cancellationToken);
        }
    }
}