lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttWriteLiveTests.cs

using System.Buffers;
using System.Collections.Concurrent;
using System.Text;
using MQTTnet;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;

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

/// <summary>
///     Live validation of MQTT write-through (Gitea #508) against the real Mosquitto fixture: a Plain
///     publish to the tag's write topic, and Sparkplug NCMD/DCMD commands, observed on the broker by an
///     independent subscriber.
/// </summary>
/// <remarks>
///     <para>
///     <b>Observed on the broker, not on a fake.</b> The whole risk this feature carries is a write that
///     reports Good and reaches nothing, so the assertion has to be that a real message arrived on a
///     real broker at the topic the operator authored. A test double asserting "publish was called"
///     could not distinguish the feature working from the feature publishing to the wrong topic.
///     </para>
///     <para>
///     <b>Env-gated + skip-clean</b>, like every suite here: without <c>MQTT_FIXTURE_ENDPOINT</c> and
///     credentials each test calls <c>Assert.Skip</c>. See <see cref="MqttFixture"/>.
///     </para>
/// </remarks>
[Collection(MqttFixtureCollection.Name)]
[Trait("Category", "LiveIntegration")]
public sealed class MqttWriteLiveTests(MqttFixture fixture)
{
    private const uint Good = 0x00000000u;
    private const uint BadNotWritable = 0x803B0000u;

    private static readonly TimeSpan MessageTimeout = TimeSpan.FromSeconds(30);

    private readonly MqttFixture _fx = fixture;

    private void SkipIfUnconfigured() => Assert.SkipWhen(_fx.SkipReason is not null, _fx.SkipReason ?? "");

    // -----------------------------------------------------------------------
    // Plain
    // -----------------------------------------------------------------------

    [Fact]
    public async Task Plain_write_publishes_the_value_on_the_authored_write_topic()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        const string rawPath = "Fixture/Mqtt/Line1/Setpoint";
        var readTopic = _fx.Topic($"write/{Guid.NewGuid():N}/state");
        var writeTopic = _fx.Topic($"write/{Guid.NewGuid():N}/set");

        await using var watcher = await TopicWatcher.StartAsync(_fx, writeTopic, ct);

        var tag = new RawTagEntry(
            rawPath,
            $$"""{"topic":"{{readTopic}}","writeTopic":"{{writeTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
            WriteIdempotent: false);

        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write");
        await driver.InitializeAsync(string.Empty, ct);

        var results = await driver.WriteAsync([new WriteRequest(rawPath, 42.5)], ct);

        results.ShouldHaveSingleItem().StatusCode.ShouldBe(Good);

        var payload = await watcher.WaitForOneAsync(MessageTimeout, ct);
        Encoding.UTF8.GetString(payload).ShouldBe(
            "42.5",
            "the write must arrive on the authored write topic carrying the value, in invariant form");
    }

    [Fact]
    public async Task Plain_write_never_lands_on_the_tags_read_topic()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        const string rawPath = "Fixture/Mqtt/Line1/SetpointSplit";
        var suffix = Guid.NewGuid().ToString("N");
        var readTopic = _fx.Topic($"write/{suffix}/state");
        var writeTopic = _fx.Topic($"write/{suffix}/set");

        // Both watchers are live BEFORE the write, so "nothing on the read topic" is a real observation
        // over the same window rather than an absence that was never looked for.
        await using var readWatcher = await TopicWatcher.StartAsync(_fx, readTopic, ct);
        await using var writeWatcher = await TopicWatcher.StartAsync(_fx, writeTopic, ct);

        var tag = new RawTagEntry(
            rawPath,
            $$"""{"topic":"{{readTopic}}","writeTopic":"{{writeTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
            WriteIdempotent: false);

        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write-split");
        await driver.InitializeAsync(string.Empty, ct);

        (await driver.WriteAsync([new WriteRequest(rawPath, 7.25)], ct))[0].StatusCode.ShouldBe(Good);

        // Positive control first: once the write HAS been seen, the read topic has had its chance.
        await writeWatcher.WaitForOneAsync(MessageTimeout, ct);
        readWatcher.Count.ShouldBe(
            0,
            "a write echoed onto the tag's own read topic would read back as a confirmed device value");
    }

    [Fact]
    public async Task Plain_write_to_a_tag_with_no_write_topic_is_refused_end_to_end()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        const string rawPath = "Fixture/Mqtt/Line1/ReadOnly";
        var tag = new RawTagEntry(
            rawPath,
            $$"""{"topic":"{{_fx.Topic("line1/temperature")}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
            WriteIdempotent: false);

        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write-readonly");
        await driver.InitializeAsync(string.Empty, ct);

        // Connected and resolvable — so BadNotWritable here is specifically "you never said where",
        // not a connection failure wearing the same status.
        (await driver.WriteAsync([new WriteRequest(rawPath, 1.0)], ct))[0].StatusCode.ShouldBe(BadNotWritable);
    }

    // -----------------------------------------------------------------------
    // Sparkplug B
    // -----------------------------------------------------------------------

    [Fact]
    public async Task Sparkplug_write_to_a_device_metric_publishes_a_DCMD()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = $"WriteGate{Guid.NewGuid():N}"[..20];
        const string node = "EdgeW";
        const string device = "Filler1";
        const string metric = "Setpoint";
        const string rawPath = "Fixture/Mqtt/Sp/Setpoint";

        var topic = $"spBv1.0/{group}/DCMD/{node}/{device}";
        await using var watcher = await TopicWatcher.StartAsync(_fx, topic, ct);

        var tag = new RawTagEntry(
            rawPath,
            $$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","deviceId":"{{device}}","metricName":"{{metric}}","dataType":"Float64"}""",
            WriteIdempotent: false);

        await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-dcmd");
        await driver.InitializeAsync(string.Empty, ct);

        (await driver.WriteAsync([new WriteRequest(rawPath, 12.5)], ct))[0].StatusCode.ShouldBe(Good);

        var bytes = await watcher.WaitForOneAsync(MessageTimeout, ct);
        var payload = Payload.Parser.ParseFrom(bytes);

        payload.Metrics.ShouldHaveSingleItem();
        payload.Metrics[0].Name.ShouldBe(metric, "a command must address the metric by NAME, never by alias");
        payload.Metrics[0].Datatype.ShouldBe((uint)TahuDataType.Double);
        payload.Metrics[0].DoubleValue.ShouldBe(12.5);
        payload.HasSeq.ShouldBeFalse("seq belongs to the edge node's own outbound stream, not to a command");
    }

    [Fact]
    public async Task Sparkplug_write_to_an_edge_node_metric_publishes_an_NCMD()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = $"WriteGate{Guid.NewGuid():N}"[..20];
        const string node = "EdgeW";
        const string metric = "NodeSetpoint";
        const string rawPath = "Fixture/Mqtt/Sp/NodeSetpoint";

        await using var watcher = await TopicWatcher.StartAsync(_fx, $"spBv1.0/{group}/NCMD/{node}", ct);

        var tag = new RawTagEntry(
            rawPath,
            $$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","metricName":"{{metric}}","dataType":"Int32"}""",
            WriteIdempotent: false);

        await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-ncmd");
        await driver.InitializeAsync(string.Empty, ct);

        (await driver.WriteAsync([new WriteRequest(rawPath, 7)], ct))[0].StatusCode.ShouldBe(Good);

        // NCMD is a SHARED channel: the driver publishes its own rebirth request
        // ("Node Control/Rebirth") on this exact topic when it connects, and the live gate caught it
        // arriving first. Wait for the command carrying OUR metric rather than for the first NCMD —
        // the DCMD case has no such contention, which is why only this one needed it.
        var payload = Payload.Parser.ParseFrom(
            await watcher.WaitForMatchAsync(
                bytes => Payload.Parser.ParseFrom(bytes).Metrics.Any(m => m.Name == metric),
                MessageTimeout,
                ct));

        var written = payload.Metrics.ShouldHaveSingleItem();
        written.Name.ShouldBe(metric);
        written.Datatype.ShouldBe((uint)TahuDataType.Int32);
        written.IntValue.ShouldBe(7u);
    }

    [Fact]
    public async Task A_sparkplug_command_is_not_retained()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = $"WriteGate{Guid.NewGuid():N}"[..20];
        const string node = "EdgeW";
        const string rawPath = "Fixture/Mqtt/Sp/Retain";
        var topic = $"spBv1.0/{group}/NCMD/{node}";

        await using var firstWatcher = await TopicWatcher.StartAsync(_fx, topic, ct);

        var tag = new RawTagEntry(
            rawPath,
            $$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","metricName":"Cmd","dataType":"Boolean"}""",
            WriteIdempotent: false);

        await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-retain");
        await driver.InitializeAsync(string.Empty, ct);
        (await driver.WriteAsync([new WriteRequest(rawPath, true)], ct))[0].StatusCode.ShouldBe(Good);
        await firstWatcher.WaitForOneAsync(MessageTimeout, ct);

        // A retained command would be replayed by the broker to every LATER subscriber — including the
        // device's own reconnect — so a single stale write would be re-applied forever. Subscribing
        // fresh, after the publish, is the only way to observe that.
        await using var lateWatcher = await TopicWatcher.StartAsync(_fx, topic, ct);
        await Task.Delay(TimeSpan.FromSeconds(2), ct);

        // Counting OUR command only. The driver's own rebirth NCMD shares this topic, so a bare count
        // would make this assertion depend on rebirth timing rather than on the retain flag.
        lateWatcher.CountMatching(bytes => Payload.Parser.ParseFrom(bytes).Metrics.Any(m => m.Name == "Cmd"))
            .ShouldBe(0, "a retained NCMD/DCMD would drive a device into a permanent command loop");
    }

    // -----------------------------------------------------------------------
    // Helpers
    // -----------------------------------------------------------------------

    private MqttDriverOptions PlainOptions(params RawTagEntry[] rawTags) => new()
    {
        Host = _fx.PlainHost,
        Port = _fx.PlainPort,
        UseTls = false,
        Username = _fx.Username,
        Password = _fx.Password,
        ClientId = $"otopcua-live-{Guid.NewGuid():N}",
        ConnectTimeoutSeconds = 15,
        Mode = MqttMode.Plain,
        Plain = new MqttPlainOptions { DefaultQos = 1 },
        RawTags = rawTags,
    };

    private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new()
    {
        Host = _fx.PlainHost,
        Port = _fx.PlainPort,
        UseTls = false,
        Username = _fx.Username,
        Password = _fx.Password,
        ClientId = $"otopcua-live-{Guid.NewGuid():N}",
        ConnectTimeoutSeconds = 15,
        Mode = MqttMode.SparkplugB,
        Sparkplug = new MqttSparkplugOptions { GroupId = groupId },
        RawTags = rawTags,
    };

    /// <summary>
    ///     An independent broker subscriber that records the payloads seen on one concrete topic — the
    ///     third party that makes "the write actually reached the broker" an observation rather than an
    ///     assumption about the driver's own code.
    /// </summary>
    private sealed class TopicWatcher : IAsyncDisposable
    {
        private readonly IMqttClient _client;
        private readonly ConcurrentQueue<byte[]> _payloads = new();

        private TopicWatcher(IMqttClient client) => _client = client;

        /// <summary>How many messages have been observed on the watched topic.</summary>
        public int Count => _payloads.Count;

        public static async Task<TopicWatcher> StartAsync(MqttFixture fixture, string topic, CancellationToken ct)
        {
            var client = new MqttClientFactory().CreateMqttClient();
            var watcher = new TopicWatcher(client);

            client.ApplicationMessageReceivedAsync += args =>
            {
                watcher._payloads.Enqueue(args.ApplicationMessage.Payload.ToArray());
                return Task.CompletedTask;
            };

            var options = new MqttClientOptionsBuilder()
                .WithTcpServer(fixture.PlainHost, fixture.PlainPort)
                .WithCredentials(fixture.Username, fixture.Password)
                .WithClientId($"otopcua-live-watch-{Guid.NewGuid():N}")
                .WithTimeout(TimeSpan.FromSeconds(10))
                .Build();

            await client.ConnectAsync(options, ct).ConfigureAwait(false);
            await client.SubscribeAsync(
                new MqttClientSubscribeOptionsBuilder().WithTopicFilter(f => f.WithTopic(topic)).Build(),
                ct).ConfigureAwait(false);

            return watcher;
        }

        /// <summary>How many observed messages satisfy <paramref name="predicate"/>.</summary>
        /// <param name="predicate">The payload test.</param>
        /// <returns>The matching count.</returns>
        public int CountMatching(Func<byte[], bool> predicate) => _payloads.Count(p => Safe(predicate, p));

        /// <summary>
        ///     Waits for a message satisfying <paramref name="predicate"/>. Needed where the topic is
        ///     shared — an NCMD topic also carries the driver's own rebirth requests.
        /// </summary>
        /// <param name="predicate">The payload test.</param>
        /// <param name="timeout">Bound on the wait.</param>
        /// <param name="ct">Caller cancellation.</param>
        /// <returns>The first matching payload.</returns>
        public async Task<byte[]> WaitForMatchAsync(Func<byte[], bool> predicate, TimeSpan timeout, CancellationToken ct)
        {
            var deadline = DateTime.UtcNow + timeout;
            while (true)
            {
                if (_payloads.FirstOrDefault(p => Safe(predicate, p)) is { } match)
                {
                    return match;
                }

                if (DateTime.UtcNow >= deadline)
                {
                    throw new TimeoutException(
                        $"No matching message arrived on the watched topic within {timeout} " +
                        $"({_payloads.Count} message(s) seen, none matching).");
                }

                await Task.Delay(100, ct).ConfigureAwait(false);
            }
        }

        /// <summary>Applies a payload predicate, treating an undecodable payload as "no match".</summary>
        /// <param name="predicate">The payload test.</param>
        /// <param name="payload">The observed bytes.</param>
        /// <returns><see langword="true"/> when the payload matches.</returns>
        private static bool Safe(Func<byte[], bool> predicate, byte[] payload)
        {
            try { return predicate(payload); }
            catch { return false; }
        }

        public async Task<byte[]> WaitForOneAsync(TimeSpan timeout, CancellationToken ct)
        {
            var deadline = DateTime.UtcNow + timeout;
            while (!_payloads.TryPeek(out _))
            {
                if (DateTime.UtcNow >= deadline)
                {
                    throw new TimeoutException(
                        $"No message arrived on the watched topic within {timeout}. The write reported success " +
                        "but nothing reached the broker — the exact failure this gate exists to catch.");
                }

                await Task.Delay(100, ct).ConfigureAwait(false);
            }

            _payloads.TryDequeue(out var payload);
            return payload!;
        }

        public async ValueTask DisposeAsync()
        {
            try { await _client.DisconnectAsync(cancellationToken: default).ConfigureAwait(false); }
            catch { /* teardown must not mask a test failure */ }

            _client.Dispose();
        }
    }
}