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

using System.Buffers;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using MQTTnet;
using MQTTnet.Protocol;
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

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

/// <summary>
///     Bespoke MQTT address-picker browser: opens a short-lived, <b>strictly passive</b> observation
///     window against the broker named by the form's JSON.
///     <para>
///         MQTT has no discovery protocol, so there is nothing to enumerate — the browser subscribes
///         to a wildcard (<c>#</c> / <c>{topicPrefix}/#</c> in plain mode, <c>spBv1.0/{groupId}/#</c>
///         in Sparkplug mode) and lets <see cref="MqttBrowseSession"/> accumulate whatever arrives:
///         a topic segment tree in plain mode, a Group → EdgeNode → Device → Metric tree decoded from
///         observed birth certificates in Sparkplug mode. Nothing on any browse path publishes — in
///         either mode: an operator opening a picker must not be able to inject a message into a
///         running plant. The session's operator-triggered <c>RequestRebirthAsync</c> is the sole
///         sanctioned publish, and it is never reached by <c>OpenAsync</c> or by any browse call.
///     </para>
///     <para>
///         <b>Layering note.</b> This project depends on the shared leaves only — <c>.Contracts</c> for
///         the Sparkplug wire schema and codec, <c>.Transport</c> for
///         <see cref="MqttClientOptionsFactory"/> — and NOT on the runtime <c>.Driver</c> project, which
///         it used to reference under a documented exception (Gitea #512). Reuse of the TLS / CA-pin path
///         is what that exception bought, and it is preserved: there is still exactly one copy, now in
///         <c>.Transport</c>. Do not fork it. Pinned by <c>MqttBrowserLayeringTests</c>.
///     </para>
/// </summary>
public sealed class MqttDriverBrowser : IDriverBrowser
{
    /// <summary>Floor on the open budget — a 1 s <c>ConnectTimeoutSeconds</c> would make browse unusable.</summary>
    internal const int MinOpenBudgetSeconds = 5;

    /// <summary>Ceiling on the open budget — the picker must never hang on an unreachable broker.</summary>
    internal const int MaxOpenBudgetSeconds = 30;

    /// <summary>Marks the transient browse identity in the broker's client-id/session logs.</summary>
    internal const string BrowseClientIdPrefix = "-browse-";

    private readonly ILogger<MqttDriverBrowser> _logger;

    /// <summary>
    ///     Creates a browser. <b>Connection-free by contract</b> — the universal browser's
    ///     <c>CanBrowse</c> constructs a throwaway instance purely to ask which driver type it
    ///     handles, so the constructor must never touch the network.
    /// </summary>
    /// <param name="logger">Optional logger; defaults to <see cref="NullLogger{T}"/>. Never receives credentials.</param>
    public MqttDriverBrowser(ILogger<MqttDriverBrowser>? logger = null) =>
        _logger = logger ?? NullLogger<MqttDriverBrowser>.Instance;

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

    /// <inheritdoc />
    /// <remarks>
    ///     Connects with a browse-only client id, subscribes the wildcard at QoS 0, and hands back a
    ///     session that serves whatever the subscription observes. The whole open is bounded by
    ///     <see cref="OpenBudget"/>; nothing here publishes.
    /// </remarks>
    public async Task<IBrowseSession> OpenAsync(string configJson, CancellationToken cancellationToken)
    {
        // MqttJson.Options — the ONE shared instance across factory / probe / driver / browser
        // (see its remarks). Parsing the same DriverConfig blob through a second, divergent
        // JsonSerializerOptions is this repo's documented systemic enum bug: the picker would accept
        // a `mode` / `protocolVersion` spelling the deployed driver rejects, or vice versa.
        var opts = JsonSerializer.Deserialize<MqttDriverOptions>(configJson, MqttJson.Options)
            ?? throw new InvalidOperationException("Mqtt options deserialized to null.");

        var filter = BuildRootFilter(opts);
        var suffix = BuildBrowseClientIdSuffix();
        var clientOptions = MqttClientOptionsFactory.BuildClientOptions(ToBrowseOptions(opts), suffix, _logger);

        using var openCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        openCts.CancelAfter(OpenBudget(opts));

        var client = new MqttClientFactory().CreateMqttClient();
        var session = new MqttBrowseSession(opts.Mode, client, _logger);
        try
        {
            // Runs on MQTTnet's dispatcher: record and return, nothing else.
            client.ApplicationMessageReceivedAsync += args =>
            {
                var payload = args.ApplicationMessage.Payload;
                ReadOnlySpan<byte> body = payload.IsSingleSegment ? payload.FirstSpan : payload.ToArray();
                session.Observe(args.ApplicationMessage.Topic, body);
                return Task.CompletedTask;
            };

            await client.ConnectAsync(clientOptions, openCts.Token).ConfigureAwait(false);

            // QoS 0: a transient observation window has no delivery guarantee to offer and should
            // cost the broker as little as possible.
            var subscribeOptions = new MqttClientSubscribeOptionsBuilder()
                .WithTopicFilter(f => f
                    .WithTopic(filter)
                    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce))
                .Build();
            await client.SubscribeAsync(subscribeOptions, openCts.Token).ConfigureAwait(false);

            _logger.LogInformation(
                "AdminUI MQTT browse session opened against {Host}:{Port} in {Mode} mode observing "
                + "'{Filter}' (read-only).",
                opts.Host, opts.Port, opts.Mode, filter);

            return session;
        }
        catch
        {
            await session.DisposeAsync().ConfigureAwait(false); // owns the client — disconnects + disposes
            throw;
        }
    }

    /// <summary>
    ///     The wildcard the observation window subscribes to.
    ///     <para>
    ///         <b>Plain mode:</b> the whole broker (<c>#</c>) unless the config scopes it with a
    ///         topic prefix.
    ///     </para>
    ///     <para>
    ///         <b>Sparkplug mode:</b> the Sparkplug namespace only —
    ///         <c>spBv1.0/{groupId}/#</c>, matching the group the deployed driver itself subscribes
    ///         (design §3.1), or <c>spBv1.0/#</c> when no group has been authored yet, so the picker
    ///         can discover which groups exist. The Plain-mode topic prefix is deliberately ignored:
    ///         it is a different mode's key, and honouring it would silently narrow the window to a
    ///         prefix that cannot match a Sparkplug topic at all.
    ///     </para>
    /// </summary>
    /// <param name="options">The browse configuration.</param>
    /// <returns>The MQTT topic filter to subscribe.</returns>
    internal static string BuildRootFilter(MqttDriverOptions options)
    {
        if (options.Mode == MqttMode.SparkplugB)
        {
            var groupId = (options.Sparkplug?.GroupId ?? string.Empty).Trim();
            return groupId.Length == 0 ? "spBv1.0/#" : $"spBv1.0/{groupId}/#";
        }

        var prefix = (options.Plain?.TopicPrefix ?? string.Empty).Trim();
        if (prefix.Length == 0) return "#";
        if (!prefix.EndsWith('/')) prefix += "/";
        return prefix + "#";
    }

    /// <summary>
    ///     A unique per-session client-id suffix. A broker disconnects an existing client when a new
    ///     one CONNECTs with the <i>same</i> client id, so sharing the driver's id would knock the
    ///     live driver offline every time an operator opened the picker.
    /// </summary>
    /// <returns>The suffix appended to the configured client id.</returns>
    internal static string BuildBrowseClientIdSuffix() =>
        BrowseClientIdPrefix + Guid.NewGuid().ToString("N")[..8];

    /// <summary>
    ///     Projects the authored options into the ones the browse CONNECT uses. Clean session is
    ///     forced: a transient browse identity must not leave queued-message state behind on the
    ///     broker after the picker closes.
    ///     <para>
    ///         ⚠ <b>THE landmine — a Last Will would break the read-only guarantee from outside it.</b>
    ///         <c>MqttDriverOptions</c> (and <c>MqttSparkplugOptions</c>) carry no will, <b>re-verified
    ///         when Task 23 unsealed Sparkplug browse</b>, so there is nothing to strip and the
    ///         projection stays a single <c>CleanSession</c> override. If a will is ever added — NDEATH
    ///         <i>is</i> a will message — it MUST be cleared here: a will is published by the
    ///         <i>broker</i>, not by us, so it is invisible to
    ///         <see cref="MqttBrowseSession.PublishCountForTest"/>, and any ungraceful end to a browse
    ///         session (crash, dropped link, the disconnect deadline expiring) would then fire an
    ///         NDEATH under the plant's own edge-node identity — killing a live Sparkplug node because
    ///         an operator opened an address picker. <c>MqttBrowseSessionTests</c>'
    ///         <c>MqttDriverOptions_CarryNothingWillShaped_…</c> is the reflection guard that turns that
    ///         "there is nothing to strip" into a checked fact rather than a claim.
    ///     </para>
    /// </summary>
    /// <param name="options">The authored configuration.</param>
    /// <returns>The configuration the browse CONNECT is built from.</returns>
    internal static MqttDriverOptions ToBrowseOptions(MqttDriverOptions options) =>
        options with { CleanSession = true };

    /// <summary>
    ///     Whole-open bound (connect + subscribe), clamped to
    ///     <see cref="MinOpenBudgetSeconds"/>–<see cref="MaxOpenBudgetSeconds"/> around the config's
    ///     own connect deadline.
    /// </summary>
    /// <param name="options">The browse configuration.</param>
    /// <returns>The clamped open budget.</returns>
    internal static TimeSpan OpenBudget(MqttDriverOptions options) =>
        TimeSpan.FromSeconds(Math.Clamp(options.ConnectTimeoutSeconds, MinOpenBudgetSeconds, MaxOpenBudgetSeconds));
}