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

using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator;

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

/// <summary>
///     Live gate for structured Sparkplug metrics (Gitea #515): a <c>Template</c> and a
///     <c>DataSet</c> published by a real edge node over the Mosquitto fixture, flattened into
///     addressable OPC UA tags.
/// </summary>
/// <remarks>
///     <para>
///         <b>What this adds over the offline suite.</b> The offline tests build decoded payloads by
///         hand, so they prove the mapping but not the <i>wire</i>: that a nested template really does
///         round-trip through protobuf with its member names intact, and — the case a naive
///         implementation gets wrong — that a DDATA structured metric arrives carrying the parent's
///         <b>alias and nothing else</b>, so the member names cannot be composed until the alias table
///         has resolved the parent.
///     </para>
///     <para>
///         The issue's acceptance criterion, verbatim: "a simulator publishing a <c>Template</c> and a
///         <c>DataSet</c> metric produces addressable, correctly-typed OPC UA nodes, live-verified;
///         every still-unsupported kind continues to skip with a warning."
///     </para>
/// </remarks>
[Trait("Category", "Integration")]
[Collection("MqttFixture")]
public sealed class SparkplugStructuredLiveTests(MqttFixture fixture)
{
    private const uint Good = 0x00000000u;
    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(20);

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

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

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("tmpl");
        const string node = "EdgeA";
        const string device = "Filler1";

        const string speedPath = "Fixture/Sparkplug/Motor/Speed";
        const string runningPath = "Fixture/Sparkplug/Motor/Running";
        const string statorPath = "Fixture/Sparkplug/Motor/Stator/Temp";

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Motor", 13, DataType.Template, MotorTemplate(speed: 1500, running: true, statorTemp: 71.5)));

        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(
                group,
                Tag(speedPath, group, node, device, "Motor/Speed"),
                Tag(runningPath, group, node, device, "Motor/Running"),
                Tag(statorPath, group, node, device, "Motor/Stator/Temp")),
            "mqtt-live-sp-template");
        var recorder = new Recorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([speedPath, runningPath, statorPath], TimeSpan.FromSeconds(1), ct);

        // The DBIRTH is a full snapshot, so these are the birth's own member values — including the
        // NESTED template's, which is what proves the `/`-joined path composes past one level.
        var birth = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1500d), Timeout, ct);
        birth.Snapshot.StatusCode.ShouldBe(Good);
        await recorder.WaitAsync(runningPath, v => Equals(v.Value, true), Timeout, ct);
        await recorder.WaitAsync(statorPath, v => Equals(v.Value, 71.5d), Timeout, ct);

        // …and now DDATA. The metric on the wire carries the PARENT's alias and no name at all, so a
        // driver that flattened only at decode time — without resolving the parent alias to its name
        // first — would have nothing to compose `Motor/Speed` from and would feed none of these.
        await sim.PublishDeviceDataAsync(
            device, [("Motor", MotorTemplate(speed: 1750, running: false, statorTemp: 68.25))], ct);

        var data = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1750d), Timeout, ct);
        data.Snapshot.StatusCode.ShouldBe(Good);
        data.FullReference.ShouldBe(speedPath, "values publish under the tag's RawPath, never the metric name");

        await recorder.WaitAsync(runningPath, v => Equals(v.Value, false), Timeout, ct);
        await recorder.WaitAsync(statorPath, v => Equals(v.Value, 68.25d), Timeout, ct);

        // The read path serves the same last values under the same keys.
        var read = await driver.ReadAsync([speedPath, runningPath, statorPath], ct);
        read.Select(r => r.StatusCode).ShouldAllBe(s => s == Good);
        read[0].Value.ShouldBe(1750d);
    }

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

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("dset");
        const string node = "EdgeA";
        const string device = "Filler1";

        const string step0Path = "Fixture/Sparkplug/Recipe/Step0";
        const string target1Path = "Fixture/Sparkplug/Recipe/Target1";

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, RecipeDataSet(baseTarget: 20d)));

        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(
                group,
                // The row index is always present, including on row 0 — the naming is uniform so a
                // table that grows a row never renames the tags already authored against it.
                Tag(step0Path, group, node, device, "Recipe/Step/0"),
                Tag(target1Path, group, node, device, "Recipe/Target/1")),
            "mqtt-live-sp-dataset");
        var recorder = new Recorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([step0Path, target1Path], TimeSpan.FromSeconds(1), ct);

        // Cells are typed per COLUMN: Step is Int32, Target is Double. A dataset whose columns and
        // types disagree is refused outright, so a wrong type here is not a near-miss — it is the
        // difference between a typed tag and no tag.
        await recorder.WaitAsync(step0Path, v => Equals(v.Value, 0), Timeout, ct);
        await recorder.WaitAsync(target1Path, v => Equals(v.Value, 21d), Timeout, ct);

        await sim.PublishDeviceDataAsync(device, [("Recipe", RecipeDataSet(baseTarget: 30d))], ct);

        var target = await recorder.WaitAsync(target1Path, v => Equals(v.Value, 31d), Timeout, ct);
        target.Snapshot.StatusCode.ShouldBe(Good);
    }

    [Fact]
    public async Task A_template_definition_is_still_skipped_rather_than_flattened()
    {
        // Skip-and-warn survives — the requirement the issue is explicit about. A definition names its
        // members and carries no values, so flattening it would materialise tags that can never read,
        // indistinguishable in the picker from an instance's.
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("tdef");
        const string node = "EdgeA";
        const string device = "Filler1";

        const string speedPath = "Fixture/Sparkplug/Proto/Speed";
        const string livePath = "Fixture/Sparkplug/Proto/Live";

        var definition = new Payload.Types.Template { Version = "1", IsDefinition = true };
        definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)DataType.Double });

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetDeviceMetrics(
            device,
            new SparkplugMetricSeed("MotorType", 15, DataType.Template, definition),

            // The positive control, and it is load-bearing: without it "Speed never arrived" is
            // satisfied by any failure to connect, birth or subscribe, and the test would pass against
            // a driver that ingests nothing at all.
            new SparkplugMetricSeed("Live", 16, DataType.Double, 42.5d));

        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(
                group,
                Tag(speedPath, group, node, device, "MotorType/Speed"),
                Tag(livePath, group, node, device, "Live")),
            "mqtt-live-sp-tdef");
        var recorder = new Recorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([speedPath, livePath], TimeSpan.FromSeconds(1), ct);

        // Wait for the control FIRST, so the absence assertion below is made at a moment the birth is
        // known to have been processed.
        await recorder.WaitAsync(livePath, v => Equals(v.Value, 42.5d), Timeout, ct);

        recorder.Count(speedPath).ShouldBe(0, "a template definition carries no values to publish");

        var read = await driver.ReadAsync([speedPath], ct);
        read[0].StatusCode.ShouldNotBe(Good, "a tag bound to an unflattened member has never had a value");
    }

    // ---- helpers -----------------------------------------------------------------------------------

    private static string NewGroup(string label) => $"OtOpcUaLive-{label}-{Guid.NewGuid():N}"[..40];

    private static Payload.Types.Template MotorTemplate(double speed, bool running, double statorTemp)
    {
        var stator = new Payload.Types.Template { Version = "1", IsDefinition = false };
        stator.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = statorTemp,
        });

        var motor = new Payload.Types.Template { Version = "1", IsDefinition = false };
        motor.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = speed,
        });
        motor.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = running,
        });
        motor.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator,
        });
        return motor;
    }

    private static Payload.Types.DataSet RecipeDataSet(double baseTarget)
    {
        var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL };
        dataSet.Columns.Add("Step");
        dataSet.Columns.Add("Target");
        dataSet.Types_.Add((uint)DataType.Int32);
        dataSet.Types_.Add((uint)DataType.Double);

        for (var row = 0; row < 2; row++)
        {
            var r = new Payload.Types.DataSet.Types.Row();
            r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row });
            r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = baseTarget + row });
            dataSet.Rows.Add(r);
        }

        return dataSet;
    }

    private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new()
    {
        Host = fixture.TlsHost,
        Port = fixture.TlsPort,
        UseTls = true,
        CaCertificatePath = fixture.CaCertificatePath,
        AllowUntrustedServerCertificate = fixture.CaCertificatePath is null,
        Username = fixture.Username,
        Password = fixture.Password,
        ClientId = $"otopcua-live-spx-{Guid.NewGuid():N}",
        ConnectTimeoutSeconds = 15,
        Mode = MqttMode.SparkplugB,
        Sparkplug = new MqttSparkplugOptions { GroupId = groupId, RequestRebirthOnGap = true },
        RawTags = rawTags,
    };

    private SparkplugEdgeNodeOptions SimOptions(string groupId, string edgeNodeId) => new()
    {
        Host = fixture.TlsHost,
        Port = fixture.TlsPort,
        UseTls = true,
        CaCertificatePath = fixture.CaCertificatePath,
        AllowUntrustedServerCertificate = fixture.CaCertificatePath is null,
        Username = fixture.Username,
        Password = fixture.Password,
        GroupId = groupId,
        EdgeNodeId = edgeNodeId,
    };

    private static RawTagEntry Tag(
        string rawPath, string groupId, string edgeNodeId, string? deviceId, string metricName)
    {
        var device = deviceId is null ? "" : $"\"deviceId\":\"{deviceId}\",";
        return new RawTagEntry(
            rawPath,
            $$"""{"groupId":"{{groupId}}","edgeNodeId":"{{edgeNodeId}}",{{device}}"metricName":"{{metricName}}"}""",
            WriteIdempotent: false);
    }

    /// <summary>Records every published data change, so a test can wait on one and count others.</summary>
    private sealed class Recorder
    {
        private readonly List<DataChangeEventArgs> _events = [];
        private readonly Lock _gate = new();

        public Recorder(MqttDriver driver) => driver.OnDataChange += (_, e) =>
        {
            lock (_gate)
            {
                _events.Add(e);
            }
        };

        public int Count(string rawPath)
        {
            lock (_gate)
            {
                return _events.Count(e => e.FullReference == rawPath);
            }
        }

        public async Task<DataChangeEventArgs> WaitAsync(
            string rawPath, Func<DataValueSnapshot, bool> predicate, TimeSpan timeout, CancellationToken ct)
        {
            var deadline = DateTime.UtcNow + timeout;
            while (true)
            {
                lock (_gate)
                {
                    foreach (var candidate in _events)
                    {
                        if (candidate.FullReference == rawPath && predicate(candidate.Snapshot))
                        {
                            return candidate;
                        }
                    }
                }

                if (DateTime.UtcNow >= deadline)
                {
                    lock (_gate)
                    {
                        throw new TimeoutException(
                            $"No matching value for '{rawPath}' within {timeout}. Seen: "
                            + string.Join(", ", _events.Select(e => $"{e.FullReference}={e.Snapshot.Value}")));
                    }
                }

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