lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs

using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;

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

/// <summary>
///     Decodes Sparkplug-B wire bytes into <see cref="SparkplugPayload"/> — the driver-side projection
///     the ingest state machine consumes. Decode only; the NCMD <i>encode</i> path lives in
///     <c>RebirthRequester</c>.
/// </summary>
/// <remarks>
///     <para>
///         <b>Nothing here throws, for any input.</b> This runs on MQTTnet's shared dispatcher thread,
///         behind an unauthenticated firehose the plant's edge nodes publish into. An escaping
///         exception would not degrade one tag — it would stall or kill delivery for every
///         subscription on the connection. Garbage bytes, a truncated body, a zero-length payload, a
///         valid protobuf of some other schema and a pathologically nested Template all resolve to a
///         verdict: <see cref="TryDecode"/> returns <see langword="false"/> and
///         <see cref="Decode"/> returns <see cref="SparkplugPayload.Invalid"/>.
///     </para>
///     <para>
///         <b>Zero-length input is invalid, not empty.</b> Protobuf would happily parse zero bytes as
///         "a Payload with every field defaulted", but in Sparkplug an empty MQTT body is never a
///         legitimate message — treating it as a well-formed payload carrying no metrics is how a
///         truncated-to-nothing body gets mistaken for a real one. It is reported as invalid.
///     </para>
///     <para>
///         <b>Explicit presence is preserved, everywhere.</b> The vendored schema is proto2 precisely
///         so absence and zero stay distinguishable, and every projection below reads
///         <c>Has{Seq,Name,Alias,Datatype,Timestamp,IsNull}</c> rather than testing a value against its
///         default. Two cases make this load-bearing rather than pedantic: a Sparkplug NBIRTH is
///         REQUIRED to carry <c>seq = 0</c>, and every DATA metric after a birth carries an alias with
///         <b>no</b> name and <b>no</b> datatype. A zero-check decoder reports "no sequence" for every
///         birth and "" for every DATA metric name.
///     </para>
///     <para>
///         <b>Values are projected raw — nothing is coerced or reinterpreted here.</b> A metric's value
///         arrives as the CLR type of the wire field that carried it (see
///         <see cref="SparkplugMetric.Value"/>); coercion against the authored tag's declared
///         <c>DriverDataType</c> is the consumer's job, and follows this driver's standing rule that a
///         value which does not fit is refused rather than silently converted. The one wire-level
///         translation this type does offer is <see cref="ReinterpretSigned"/>, because Sparkplug's
///         two's-complement-in-an-unsigned-field encoding of signed integers is a property of the
///         <i>wire</i>, not of the tag.
///     </para>
///     <para>
///         <b><c>Template</c> and <c>DataSet</c> metrics decode to
///         <see cref="SparkplugValueKind.Structured"/> with their members flattened</b> into
///         <see cref="SparkplugMetric.Members"/> — a template member <c>Speed</c> of metric
///         <c>Motor</c> becomes the addressable <c>Motor/Speed</c>. See <see cref="SparkplugFlattener"/>
///         for the mapping and its bounds. Anything that cannot be flattened honestly stays
///         <see cref="SparkplugValueKind.Unsupported"/> with a null value — deliberately visible, so a
///         consumer warns about a metric it cannot serve instead of silently treating it as one that was
///         never published, and never coerces it into a guessed scalar.
///     </para>
///     <para>
///         <b>A metric's <c>PropertySet</c> is projected to <see cref="SparkplugMetric.Properties"/></b>
///         as node metadata, never as sibling values: a PropertySet describes a metric (units, ranges,
///         quality hints); it is not something the metric <i>is</i>.
///     </para>
/// </remarks>
public static class SparkplugCodec
{
    /// <summary>Decodes Sparkplug-B wire bytes, reporting success rather than throwing.</summary>
    /// <param name="wire">The MQTT message body.</param>
    /// <param name="payload">
    ///     The decoded payload on success; <see cref="SparkplugPayload.Invalid"/> otherwise. Never
    ///     <see langword="null"/>.
    /// </param>
    /// <returns>
    ///     <see langword="true"/> when the bytes parsed as a Sparkplug-B <c>Payload</c>. Protobuf is a
    ///     permissive, self-describing-only-by-convention format, so this means "parsed" rather than
    ///     "was genuinely produced by a Sparkplug edge node" — unknown fields are preserved by
    ///     Google.Protobuf and simply ignored here.
    /// </returns>
    public static bool TryDecode(ReadOnlySpan<byte> wire, out SparkplugPayload payload)
    {
        payload = SparkplugPayload.Invalid;

        if (wire.IsEmpty)
        {
            return false;
        }

        try
        {
            var proto = Payload.Parser.ParseFrom(wire);

            var metrics = proto.Metrics.Count == 0
                ? []
                : new SparkplugMetric[proto.Metrics.Count];

            for (var i = 0; i < proto.Metrics.Count; i++)
            {
                metrics[i] = ProjectMetric(proto.Metrics[i]);
            }

            payload = new SparkplugPayload(
                IsValid: true,
                Seq: proto.HasSeq ? proto.Seq : null,
                TimestampMs: proto.HasTimestamp ? proto.Timestamp : null,
                Metrics: metrics);

            return true;
        }
        catch (Exception)
        {
            // Deliberately broad, and deliberately spanning the projection as well as the parse.
            // InvalidProtocolBufferException covers malformed, truncated and over-nested input, but
            // this is the dispatcher-thread boundary and the projection walks an attacker-shaped
            // object graph: the contract is a verdict for EVERY input, so a guard that stopped at
            // the parse would be a contract with a hole in it. Narrowing the catch would trade the
            // guarantee for a taxonomy nobody downstream can act on.
            payload = SparkplugPayload.Invalid;
            return false;
        }
    }

    /// <summary>
    ///     Decodes Sparkplug-B wire bytes, returning <see cref="SparkplugPayload.Invalid"/> when they
    ///     cannot be decoded.
    /// </summary>
    /// <param name="wire">The MQTT message body.</param>
    /// <returns>The decoded payload — never <see langword="null"/>.</returns>
    /// <remarks>
    ///     The convenience shape. <b>Check <see cref="SparkplugPayload.IsValid"/>:</b> an undecodable
    ///     body and a genuinely metric-less one both present as an empty
    ///     <see cref="SparkplugPayload.Metrics"/>, and only the flag tells them apart. Prefer
    ///     <see cref="TryDecode"/> where the verdict drives control flow.
    /// </remarks>
    public static SparkplugPayload Decode(ReadOnlySpan<byte> wire) =>
        TryDecode(wire, out var payload) ? payload : SparkplugPayload.Invalid;

    /// <summary>
    ///     Reinterprets a raw metric value as the signed integer Sparkplug encoded it as, when its
    ///     datatype is a signed integer type; returns the value unchanged for every other datatype.
    /// </summary>
    /// <param name="value">
    ///     A <see cref="SparkplugMetric.Value"/> — a <see cref="uint"/> for the 8/16/32-bit arms, a
    ///     <see cref="ulong"/> for the 64-bit arm.
    /// </param>
    /// <param name="datatype">The metric's Sparkplug datatype, from its birth or its own field.</param>
    /// <returns>
    ///     <see cref="sbyte"/> / <see cref="short"/> / <see cref="int"/> / <see cref="long"/> for the
    ///     signed integer datatypes; <paramref name="value"/> untouched otherwise (including when it is
    ///     <see langword="null"/> or not the wire type the datatype implies).
    /// </returns>
    /// <remarks>
    ///     Sparkplug carries <c>Int8</c>/<c>Int16</c>/<c>Int32</c> in the <b>unsigned</b>
    ///     <c>int_value</c> field and <c>Int64</c> in the unsigned <c>long_value</c> field, as two's
    ///     complement. Handing the raw field to a consumer publishes <c>4294967254</c> for a tag whose
    ///     value is <c>-42</c> — a wrong value that looks entirely plausible, arrives with Good
    ///     quality, and is invisible until someone reads a gauge. This is the single call that undoes
    ///     it, and it is separate from the decode itself because a DATA metric carries no datatype of
    ///     its own: only the consumer, holding the alias table built from the birth, knows which
    ///     datatype applies.
    /// </remarks>
    public static object? ReinterpretSigned(object? value, TahuDataType datatype) => datatype switch
    {
        TahuDataType.Int8 when value is uint raw => unchecked((sbyte)raw),
        TahuDataType.Int16 when value is uint raw => unchecked((short)raw),
        TahuDataType.Int32 when value is uint raw => unchecked((int)raw),
        TahuDataType.Int64 when value is ulong raw => unchecked((long)raw),
        _ => value,
    };

    /// <summary>Projects one generated metric onto the driver-side shape, presence intact.</summary>
    /// <param name="metric">The generated metric.</param>
    /// <returns>The projection.</returns>
    private static SparkplugMetric ProjectMetric(Payload.Types.Metric metric)
    {
        var (kind, value, members) = ProjectValue(metric);

        return new SparkplugMetric(
            Name: metric.HasName ? metric.Name : null,
            Alias: metric.HasAlias ? metric.Alias : null,

            // Cast, never filter: an index this build's enum does not define is a future Sparkplug
            // revision or a misbehaving publisher, and dropping it would deny the mapping layer the
            // only evidence it has.
            DataType: metric.HasDatatype ? (TahuDataType)metric.Datatype : null,
            TimestampMs: metric.HasTimestamp ? metric.Timestamp : null,
            ValueKind: kind,
            Value: value,
            Members: members,
            Properties: SparkplugFlattener.ProjectProperties(metric.Properties));
    }

    /// <summary>
    ///     Resolves a <b>template member's</b> value oneof. Internal to the assembly: shared with
    ///     <see cref="SparkplugFlattener"/>, which walks nested templates and needs the identical
    ///     projection so a member and a top-level metric of the same type never decode differently.
    /// </summary>
    /// <param name="metric">The template member, itself a generated <c>Metric</c>.</param>
    /// <returns>The value kind and the projected raw value.</returns>
    internal static (SparkplugValueKind Kind, object? Value) ProjectMemberValue(Payload.Types.Metric metric)
    {
        var (kind, value, _) = ProjectValue(metric);
        return (kind, value);
    }

    /// <summary>Resolves a metric's value <c>oneof</c> to a kind plus a raw CLR value.</summary>
    /// <param name="metric">The generated metric.</param>
    /// <returns>The value kind and the projected value.</returns>
    /// <remarks>
    ///     <c>is_null</c> wins over the <c>oneof</c>: the Sparkplug spec's whole reason for the field is
    ///     that some datatypes have no spare sentinel, so a publisher that sets it means "null" even if
    ///     a value field is also populated.
    /// </remarks>
    private static (SparkplugValueKind Kind, object? Value, IReadOnlyList<SparkplugMember> Members) ProjectValue(
        Payload.Types.Metric metric)
    {
        if (metric.HasIsNull && metric.IsNull)
        {
            return (SparkplugValueKind.Null, null, []);
        }

        // structured values flatten to addressable members. A refusal to flatten falls
        // through to Unsupported — the previous behaviour — rather than to a guessed scalar.
        switch (metric.ValueCase)
        {
            case Payload.Types.Metric.ValueOneofCase.TemplateValue:
                return SparkplugFlattener.TryFlattenTemplate(metric.TemplateValue, out var templateMembers)
                    ? (SparkplugValueKind.Structured, null, templateMembers)
                    : (SparkplugValueKind.Unsupported, null, []);

            case Payload.Types.Metric.ValueOneofCase.DatasetValue:
                return SparkplugFlattener.TryFlattenDataSet(metric.DatasetValue, out var cells)
                    ? (SparkplugValueKind.Structured, null, cells)
                    : (SparkplugValueKind.Unsupported, null, []);
        }

        return metric.ValueCase switch
        {
            Payload.Types.Metric.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, metric.IntValue, Empty),
            Payload.Types.Metric.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, metric.LongValue, Empty),
            Payload.Types.Metric.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, metric.FloatValue, Empty),
            Payload.Types.Metric.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, metric.DoubleValue, Empty),
            Payload.Types.Metric.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, metric.BooleanValue, Empty),
            Payload.Types.Metric.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, metric.StringValue, Empty),

            // Copied out of the ByteString: the projection must outlive the parsed message.
            Payload.Types.Metric.ValueOneofCase.BytesValue =>
                (SparkplugValueKind.Scalar, metric.BytesValue.ToByteArray(), Empty),

            // An extension value is a third-party escape hatch with no defined semantics — there is
            // nothing honest to flatten it into.
            Payload.Types.Metric.ValueOneofCase.ExtensionValue => (SparkplugValueKind.Unsupported, null, Empty),

            _ => (SparkplugValueKind.Absent, null, Empty),
        };
    }

    /// <summary>The shared empty member list — every scalar metric would otherwise allocate one.</summary>
    private static readonly IReadOnlyList<SparkplugMember> Empty = [];
}

/// <summary>
///     A decoded Sparkplug-B payload: the driver-side projection of the generated <c>Payload</c>,
///     carrying only what the ingest state machine consumes.
/// </summary>
/// <param name="IsValid">
///     <see langword="false"/> for a payload that could not be decoded. An invalid payload also has an
///     empty <paramref name="Metrics"/> list, so <b>this flag is the only thing distinguishing an
///     undecodable body from a legitimately metric-less one</b>.
/// </param>
/// <param name="Seq">
///     The payload sequence number, or <see langword="null"/> when the message carried none. Kept as
///     the wire's <see cref="ulong"/> rather than narrowed to a byte: the Sparkplug range is 0–255, but
///     a publisher that violates it is reporting a fact the consumer should be able to see and reject,
///     not one this layer should silently truncate into a plausible-looking sequence number.
/// </param>
/// <param name="TimestampMs">The payload timestamp in Sparkplug epoch milliseconds, or null when absent.</param>
/// <param name="Metrics">The payload's metrics, in wire order. Never <see langword="null"/>.</param>
public sealed record SparkplugPayload(
    bool IsValid,
    ulong? Seq,
    ulong? TimestampMs,
    IReadOnlyList<SparkplugMetric> Metrics)
{
    /// <summary>The shared "could not decode this" result.</summary>
    public static readonly SparkplugPayload Invalid = new(false, null, null, []);
}

/// <summary>One decoded Sparkplug metric.</summary>
/// <param name="Name">
///     The metric's stable name, or <see langword="null"/> when the message omitted it — which every
///     real DATA metric after a birth does, carrying only <paramref name="Alias"/>. Distinguishing
///     this from an empty string is the reason the vendored schema is proto2.
/// </param>
/// <param name="Alias">The per-birth alias, or <see langword="null"/> when the metric carried none.</param>
/// <param name="DataType">
///     The metric's declared datatype, or <see langword="null"/> when absent (again, the normal case
///     for a DATA metric — its datatype comes from the birth). An index this build's enum does not
///     define is preserved as an undefined enum value rather than dropped.
/// </param>
/// <param name="TimestampMs">
///     The metric's own acquisition timestamp in Sparkplug epoch milliseconds, or
///     <see langword="null"/> when it carried none — in which case the payload's timestamp applies.
/// </param>
/// <param name="ValueKind">
///     What <paramref name="Value"/> means. Check this before reading the value: a null value is
///     ambiguous between "explicitly null", "absent" and "a kind v1 does not support".
/// </param>
/// <param name="Value">
///     The <b>raw</b> wire value, boxed: <see cref="uint"/> (<c>int_value</c>), <see cref="ulong"/>
///     (<c>long_value</c>), <see cref="float"/>, <see cref="double"/>, <see cref="bool"/>,
///     <see cref="string"/> or <see cref="byte"/><c>[]</c> — and <see langword="null"/> for every
///     <paramref name="ValueKind"/> other than <see cref="SparkplugValueKind.Scalar"/>.
///     <para>
///         <b>Signed integers arrive as their unsigned two's-complement wire value</b> — a
///         <c>DataType.Int32</c> metric holding -42 is a <see cref="uint"/> of 4294967254 here. Run it
///         through <see cref="SparkplugCodec.ReinterpretSigned"/> with the metric's datatype (from the
///         birth, for a DATA metric) before publishing it.
///     </para>
///     <para>
///         Boxing is deliberate: the consumer coerces against the authored tag's declared
///         <c>DriverDataType</c> and hands the result to a <c>DataValueSnapshot</c>, which boxes
///         anyway, so a discriminated-union shape would buy an unboxing hop and cost every consumer a
///         switch over a dozen arms.
///     </para>
/// </param>
/// <param name="Members">
///     The flattened members of a <see cref="SparkplugValueKind.Structured"/> metric — a
///     <c>Template</c> or <c>DataSet</c> broken into addressable scalars. Empty for every
///     other kind. Paths are <b>relative to this metric</b>; see <see cref="SparkplugMember.Path"/>.
/// </param>
/// <param name="Properties">
///     The metric's <c>PropertySet</c>, projected as node metadata. Empty when the metric
///     carried none. <b>Never a value</b> — a PropertySet describes the metric rather than being one.
/// </param>
public readonly record struct SparkplugMetric(
    string? Name,
    ulong? Alias,
    TahuDataType? DataType,
    ulong? TimestampMs,
    SparkplugValueKind ValueKind,
    object? Value,
    IReadOnlyList<SparkplugMember>? Members = null,
    IReadOnlyList<SparkplugProperty>? Properties = null)
{
    /// <summary>The flattened members, never null.</summary>
    public IReadOnlyList<SparkplugMember> MemberList => Members ?? [];

    /// <summary>The projected properties, never null.</summary>
    public IReadOnlyList<SparkplugProperty> PropertyList => Properties ?? [];
}

/// <summary>What a decoded metric's <see cref="SparkplugMetric.Value"/> represents.</summary>
/// <remarks>
///     Three of the four members have a null <see cref="SparkplugMetric.Value"/> and mean entirely
///     different things, which is exactly why the distinction is carried explicitly rather than left
///     for a consumer to infer from a null.
/// </remarks>
public enum SparkplugValueKind
{
    /// <summary>The metric's value <c>oneof</c> carried nothing at all.</summary>
    Absent = 0,

    /// <summary>The metric set <c>is_null</c>: it exists, and its value is explicitly null.</summary>
    Null,

    /// <summary><see cref="SparkplugMetric.Value"/> holds the raw wire value.</summary>
    Scalar,

    /// <summary>
    ///     An extension value, or a <c>Template</c>/<c>DataSet</c> that could not be flattened
    ///     honestly (see <see cref="SparkplugFlattener"/>). Decoded and reported, but not servable:
    ///     the consumer warns and skips the metric rather than treating it as missing — and never
    ///     coerces it into a guessed scalar.
    /// </summary>
    Unsupported,

    /// <summary>
    ///     A <c>Template</c> or <c>DataSet</c> whose members were flattened into
    ///     <see cref="SparkplugMetric.Members"/>. <see cref="SparkplugMetric.Value"/> is
    ///     null — the parent has no scalar value of its own; its members carry the values.
    /// </summary>
    Structured,
}