lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/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>DataSet</c> and <c>Template</c> metrics are out of scope for v1.</b> They decode to
///         <see cref="SparkplugValueKind.Unsupported"/> with a null value — deliberately visible, so a
///         consumer can warn about a metric it cannot serve instead of silently treating it as one
///         that was never published.
///     </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;
        }

        Payload proto;
        try
        {
            proto = Payload.Parser.ParseFrom(wire);
        }
        catch (Exception)
        {
            // Deliberately broad. InvalidProtocolBufferException covers malformed, truncated and
            // over-nested input, but this is the dispatcher-thread boundary: the contract is a
            // verdict for EVERY input, and narrowing the catch would trade that guarantee for a
            // taxonomy nobody downstream can act on.
            return false;
        }

        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;
    }

    /// <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) = 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);
    }

    /// <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) ProjectValue(Payload.Types.Metric metric)
    {
        if (metric.HasIsNull && metric.IsNull)
        {
            return (SparkplugValueKind.Null, null);
        }

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

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

            // v1 scope boundary — decoded as a visible refusal, not as an exception or a silent drop.
            Payload.Types.Metric.ValueOneofCase.DatasetValue => (SparkplugValueKind.Unsupported, null),
            Payload.Types.Metric.ValueOneofCase.TemplateValue => (SparkplugValueKind.Unsupported, null),
            Payload.Types.Metric.ValueOneofCase.ExtensionValue => (SparkplugValueKind.Unsupported, null),

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

/// <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>
public readonly record struct SparkplugMetric(
    string? Name,
    ulong? Alias,
    TahuDataType? DataType,
    ulong? TimestampMs,
    SparkplugValueKind ValueKind,
    object? Value);

/// <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>
    ///     A <c>DataSet</c>, <c>Template</c> or extension value — decoded and reported, but not
    ///     supported in v1. The consumer should warn and skip the metric rather than treat it as
    ///     missing.
    /// </summary>
    Unsupported,
}