using System.Globalization;
using System.Text;
using System.Text.Json;
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
///
/// Why a write could not be encoded. Maps 1:1 onto the status the driver returns, so the reason a
/// write was refused is decided here — once, purely — rather than at the transport.
///
public enum MqttWriteEncodeError
{
/// Encoding succeeded.
None = 0,
///
/// The tag declares no usable write destination: a Plain tag with no writeTopic or
/// one carrying an MQTT wildcard, or a Sparkplug tag whose binding tuple is incomplete or whose
/// group/edge-node/device id is not a legal topic segment (see ).
/// Deliberately a hard refusal — see 's remarks on why a silent
/// no-op would be worse than the honest read-only behaviour this replaced.
///
NoWriteDestination,
/// The value is null, or cannot be represented in the tag's declared data type.
ValueNotRepresentable,
/// The tag's declared data type has no Sparkplug command encoding.
UnsupportedDataType,
}
///
/// Encodes an outbound MQTT write — the payload bytes and the topic to publish them on — for both
/// ingest shapes (Gitea #508). Pure: no I/O, no transport, no clock beyond the Sparkplug command
/// timestamp, so the whole encode contract is unit-testable without a broker.
///
///
///
/// A missing write destination is a refusal, never a no-op. Before #508 the driver
/// implemented no IWritable at all, so every MQTT node materialised read-only — honest, if
/// limited. The one outcome that would be strictly worse is a node that looks writable,
/// accepts a write, returns Good and publishes nothing. So a Plain tag with no writeTopic
/// yields and the driver surfaces
/// BadNotWritable.
///
///
/// The write topic is deliberately separate from the read topic. A tag legitimately reads
/// .../state and writes .../set, so the destination is its own authored field rather
/// than being inferred from . Publishing a command back onto
/// the tag's own read topic would, on a broker with no device behind it, echo straight back and
/// read as a confirmed device value.
///
///
/// Sparkplug writes are commands, not data. A metric under a device encodes as a DCMD, one
/// owned by the edge node itself as an NCMD, addressed by the tag's stable
/// (group, edgeNode, device?, metric) tuple — never by the per-birth alias, which may be
/// reused for a different metric across a rebirth. No seq is stamped, for the reason
/// documents at length: the sequence stream belongs to the messages
/// an edge node itself publishes, and a strict node validating a host-supplied one against its own
/// counter would have grounds to reject the command.
///
///
/// jsonPath is not applied on write. It is a read-side extraction from the read topic's
/// document, and the write topic is a different topic whose payload shape nothing in the tag config
/// describes. A write therefore emits the value as a bare JSON
/// literal. Reconstructing a wrapped document from an extraction path would be guesswork, and
/// guessing a payload shape is how a write lands somewhere nobody authored.
///
///
public static class MqttWriteEncoder
{
/// QoS for an outbound write. See the remarks on .
public const int WriteQos = 0;
///
/// Encodes one write for the tag under the driver's .
///
/// The driver's ingest shape — chosen by driver config, never sniffed from the tag.
/// The resolved tag definition.
/// The value to write.
/// The topic + payload to publish, when this returns .
/// The encode outcome.
///
/// QoS 0, retain . Retain is the one that matters: a retained
/// command would be replayed by the broker to the device on every future subscribe — including
/// the device's own reconnect — so a single stale write would be re-applied forever. This is the
/// same hazard avoids for NCMD, and the reason neither is
/// configurable.
///
public static MqttWriteEncodeError TryEncode(
MqttMode mode,
MqttTagDefinition def,
object? value,
out MqttEncodedWrite encoded)
{
ArgumentNullException.ThrowIfNull(def);
encoded = default;
return mode == MqttMode.SparkplugB
? TryEncodeSparkplug(def, value, out encoded)
: TryEncodePlain(def, value, out encoded);
}
/// Encodes a Plain-mode write: the value's text/JSON rendering onto the tag's write topic.
/// The resolved tag definition.
/// The value to write.
/// The topic + payload on success.
/// The encode outcome.
private static MqttWriteEncodeError TryEncodePlain(
MqttTagDefinition def,
object? value,
out MqttEncodedWrite encoded)
{
encoded = default;
if (string.IsNullOrWhiteSpace(def.WriteTopic))
{
return MqttWriteEncodeError.NoWriteDestination;
}
// A wildcard topic is not a destination — MQTT-3.3.2-2 forbids one in a PUBLISH, and a
// conformant broker answers a protocol violation by CLOSING the connection, which is the same
// connection every one of this driver's subscriptions lives on. So publishing anyway would turn
// one mis-authored tag into a whole-driver outage. `NoWriteDestination` (→ BadNotWritable) is the
// right verdict rather than a new error: the tag genuinely has nowhere a write could be sent, and
// the read path is untouched. See MqttAddressRules.
if (MqttAddressRules.HasWildcard(def.WriteTopic))
{
return MqttWriteEncodeError.NoWriteDestination;
}
if (value is null)
{
// No MQTT payload means "the value is null" — a zero-length payload is a legitimate
// clear-the-retained-message instruction, which is a different operation entirely.
return MqttWriteEncodeError.ValueNotRepresentable;
}
string text;
if (def.PayloadFormat == MqttPayloadFormat.Json)
{
try
{
text = JsonSerializer.Serialize(value);
}
catch (NotSupportedException)
{
return MqttWriteEncodeError.ValueNotRepresentable;
}
}
else
{
// Raw and Scalar are both "the payload's UTF-8 text" on the read side; the write side
// mirrors that. Invariant culture, or a comma decimal separator on a European host would
// publish a number the device cannot parse.
text = value is IFormattable f
? f.ToString(null, CultureInfo.InvariantCulture)
: value.ToString() ?? "";
}
encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text));
return MqttWriteEncodeError.None;
}
/// Encodes a Sparkplug-mode write as an NCMD (node metric) or DCMD (device metric).
/// The resolved tag definition.
/// The value to write.
/// The topic + payload on success.
/// The encode outcome.
private static MqttWriteEncodeError TryEncodeSparkplug(
MqttTagDefinition def,
object? value,
out MqttEncodedWrite encoded)
{
encoded = default;
if (def.GroupId is not { Length: > 0 } groupId
|| def.EdgeNodeId is not { Length: > 0 } edgeNodeId
|| def.MetricName is not { Length: > 0 } metricName)
{
// A Sparkplug tag's command address IS its binding tuple, so an incomplete tuple is the
// Sparkplug equivalent of a missing write topic rather than a different kind of failure.
return MqttWriteEncodeError.NoWriteDestination;
}
// …and an id that is not a legal topic SEGMENT is the same failure by a different route: the
// tuple is assembled into the NCMD/DCMD topic, so a '/' would silently address a different node
// than the tag names while '+'/'#' would make the PUBLISH a protocol violation (see the Plain
// branch). `metricName` is exempt on purpose — it rides in the payload, and Sparkplug's own
// canonical names use '/'.
if (MqttAddressRules.IsIllegalTopicSegment(groupId)
|| MqttAddressRules.IsIllegalTopicSegment(edgeNodeId)
|| MqttAddressRules.IsIllegalTopicSegment(def.DeviceId))
{
return MqttWriteEncodeError.NoWriteDestination;
}
if (value is null)
{
return MqttWriteEncodeError.ValueNotRepresentable;
}
var metric = new Payload.Types.Metric { Name = metricName };
var error = ApplyValue(metric, def.DataType, value);
if (error != MqttWriteEncodeError.None)
{
return error;
}
var timestampMs = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
metric.Timestamp = timestampMs;
var payload = new Payload { Timestamp = timestampMs };
payload.Metrics.Add(metric);
// A device metric is commanded with DCMD addressed to that device; an edge-node metric with
// NCMD. Sending a device metric as NCMD would address the node, which does not own it.
var isDevice = def.DeviceId is { Length: > 0 };
var topic = SparkplugTopic.Format(
groupId,
isDevice ? SparkplugMessageType.DCMD : SparkplugMessageType.NCMD,
edgeNodeId,
isDevice ? def.DeviceId : null);
encoded = new MqttEncodedWrite(topic, payload.ToByteArray());
return MqttWriteEncodeError.None;
}
///
/// Sets the metric's typed value field and its Datatype from the tag's declared type.
///
/// The metric being built.
/// The tag's declared driver data type.
/// The value to convert.
/// The encode outcome.
///
/// Datatype is set explicitly rather than left for the node to infer, for the same reason
/// sets it: a node that treats a missing or ambiguous field as
/// malformed drops the command silently, and a silently-dropped write is the failure mode this
/// whole feature must not have.
///
private static MqttWriteEncodeError ApplyValue(
Payload.Types.Metric metric,
DriverDataType dataType,
object value)
{
try
{
switch (dataType)
{
case DriverDataType.Boolean:
metric.Datatype = (uint)TahuDataType.Boolean;
metric.BooleanValue = Convert.ToBoolean(value, CultureInfo.InvariantCulture);
return MqttWriteEncodeError.None;
case DriverDataType.Int16:
metric.Datatype = (uint)TahuDataType.Int16;
metric.IntValue = unchecked((uint)Convert.ToInt16(value, CultureInfo.InvariantCulture));
return MqttWriteEncodeError.None;
case DriverDataType.Int32:
metric.Datatype = (uint)TahuDataType.Int32;
metric.IntValue = unchecked((uint)Convert.ToInt32(value, CultureInfo.InvariantCulture));
return MqttWriteEncodeError.None;
case DriverDataType.Int64:
metric.Datatype = (uint)TahuDataType.Int64;
metric.LongValue = unchecked((ulong)Convert.ToInt64(value, CultureInfo.InvariantCulture));
return MqttWriteEncodeError.None;
case DriverDataType.UInt16:
metric.Datatype = (uint)TahuDataType.Uint16;
metric.IntValue = Convert.ToUInt16(value, CultureInfo.InvariantCulture);
return MqttWriteEncodeError.None;
case DriverDataType.UInt32:
metric.Datatype = (uint)TahuDataType.Uint32;
metric.IntValue = Convert.ToUInt32(value, CultureInfo.InvariantCulture);
return MqttWriteEncodeError.None;
case DriverDataType.UInt64:
metric.Datatype = (uint)TahuDataType.Uint64;
metric.LongValue = Convert.ToUInt64(value, CultureInfo.InvariantCulture);
return MqttWriteEncodeError.None;
case DriverDataType.Float32:
metric.Datatype = (uint)TahuDataType.Float;
metric.FloatValue = Convert.ToSingle(value, CultureInfo.InvariantCulture);
return MqttWriteEncodeError.None;
case DriverDataType.Float64:
metric.Datatype = (uint)TahuDataType.Double;
metric.DoubleValue = Convert.ToDouble(value, CultureInfo.InvariantCulture);
return MqttWriteEncodeError.None;
case DriverDataType.String:
metric.Datatype = (uint)TahuDataType.String;
metric.StringValue = value as string ?? Convert.ToString(value, CultureInfo.InvariantCulture) ?? "";
return MqttWriteEncodeError.None;
case DriverDataType.DateTime:
metric.Datatype = (uint)TahuDataType.DateTime;
metric.LongValue = (ulong)new DateTimeOffset(
Convert.ToDateTime(value, CultureInfo.InvariantCulture).ToUniversalTime())
.ToUnixTimeMilliseconds();
return MqttWriteEncodeError.None;
default:
// DriverDataType.Reference and anything added later: refuse rather than guess a
// wire representation. An unsupported type must not become a String on the wire.
return MqttWriteEncodeError.UnsupportedDataType;
}
}
catch (Exception ex) when (ex is InvalidCastException or FormatException or OverflowException)
{
return MqttWriteEncodeError.ValueNotRepresentable;
}
}
}
/// One encoded outbound write: where it goes and what it carries.
/// The concrete topic to publish on.
/// The encoded message body.
public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload);