Measurements and online availability repeat at five-minute intervals.
The device contact timeout is fifteen minutes. These constants do not
run timers or implement publication suppression.
+
+********************
+ Typed publications
+********************
+
+Each operation has a ``Build`` function and a ``Parse`` function, such
+as ``BuildReported`` and ``ParseReported``. Builders accept an ID, an
+optional target, metadata, and a typed payload. They validate the same
+contract as the parsers. ``Parse`` selects the typed payload and address
+kind from an incoming topic. A caller does not split or join topics and
+does not build an envelope.
+
+A ``Publication`` contains topic, JSON bytes, QoS, and retention. It
+also supplies the expiry deadline, response topic, and correlation data
+when the operation needs them. The transport uses MQTT 5, sets the
+payload format to UTF-8 and the content type to ``application/json``,
+and derives the message expiry interval from the deadline. It rejects
+expired work before sending or processing it. The protocol package has
+no broker client or clock worker.
+
+.. list-table:: Delivery rules
+ :header-rows: 1
+
+ - - Operation
+ - QoS
+ - Retained
+
+ - - Descriptor, availability, reported, hub status
+ - 1
+ - Yes
+
+ - - Telemetry
+ - 0
+ - No
+
+ - - Events, commands, command batches, results
+ - 1
+ - No
+
+A parser accepts a live delivery of retained-topic data with its
+retained flag false. It rejects a retained event, telemetry sample,
+command, batch, or result. Telemetry uses the same complete value
+payload as a reported snapshot. The publisher keeps the retained
+reported snapshot up to date, with QoS 1, without topic string
+operations: it uses ``BuildReported`` for the same device and
+capability.
+
+**********
+ Payloads
+**********
+
+``DescriptorPayload`` requires nonempty manufacturer, model, firmware,
+and capabilities. Optional attribution has a 2000-byte limit. Each
+capability has a unique valid ID and a valid type. Both Boolean access
+flags must be present. At least one must be true. The descriptor does
+not contain house metadata.
+
+``AvailabilityPayload`` accepts ``online`` or ``offline``. A shared
+broker connection cannot have a separate last will for every device. The
+driver reports device failures and repeats online contact evidence.
+
+``ValuePayload`` requires a JSON value and preserves complete objects.
+JSON null is a present value. The wire layer does not check
+capability-specific units, ranges, settings, or physical access.
+Commands use this payload and require an expiry. The receiver checks
+ownership, access, and expiry before it acts.
+
+``EventPayload`` is an open object of event data. Every member is
+application data, not an optional wire field. Empty objects and custom
+members are valid. This supports exact event matching, physical button
+actions, and camera motion times without discarding event details.
+Unknown fields in envelopes and in closed payloads, including nested
+capabilities, batch members, and hub diagnostics, are rejected. Nested
+application values and event data are open by definition.
+
+``BatchPayload`` holds 2 through 256 ordered commands for one device.
+Each member has a capability, metadata, and a value payload. Encoding
+inserts the member schema. The outer message ID equals the first member
+ID. Member IDs and capabilities are distinct. Every member requires its
+own expiry. A malformed member rejects the complete batch. The transport
+uses the earliest member deadline and no combined response topic. Each
+member has its own result.
+
+``ResultPayload`` accepts ``accepted``, ``completed``, ``rejected``, or
+``failed``. Rejected and failed results require a nonempty reason.
+Accepted and completed results must not have a reason. The correlation
+ID must equal the command ID in the topic. A result does not replace
+physical feedback. Duplicate detection, result ordering, and physical
+operation safety belong to the driver and core.
+
+``HubStatusPayload`` contains one Zigbee hub snapshot with a matching
+hub ID, address, status, device counts, and optional pairing and
+diagnostics. Status is ``ready`` or ``unavailable``. Hub timestamps use
+UTC. Its envelope requires expiry. ``HubStatusLifetime`` calculates two
+poll intervals plus 50 seconds. The publisher refreshes status at the
+poll interval and after reconnect. It preserves diagnostic observation
+time and stale flags. Consumers reject expired snapshots and check
+source age before using a diagnostic value. Hub status is not core
+house-state input.
+
+**************
+ Verification
+**************
+
+Unit tests cover all topic and payload forms, delivery rules, invalid
+fields and schemas, batch order and bounds, correlation, and expiry
+metadata. These tests do not verify a broker, a physical device, MQTT
+acknowledgement, reconnect behavior, publication suppression, or command
+execution.
// Descriptor identifies retained device metadata.
Descriptor Kind = "descriptor"
+
// Availability identifies retained device reachability.
Availability Kind = "availability"
+
// Reported identifies a retained point value.
Reported Kind = "reported"
+
// Telemetry identifies a non-retained measurement.
Telemetry Kind = "telemetry"
+
// Event identifies a non-retained occurrence.
Event Kind = "events"
+
// Command identifies a non-retained point request.
Command Kind = "commands"
+
// CommandBatch identifies an ordered group of point requests.
CommandBatch Kind = "batch"
+
// Result identifies a command outcome.
Result Kind = "results"
+
// HubStatus identifies retained Zigbee hub diagnostics.
HubStatus Kind = "status"
)
switch {
case len(parts) == 6 && parts[2] == "zigbee" && parts[3] == "hubs" && parts[5] == "status":
- address = Address{Kind: HubStatus, ID: parts[4]}
+ address = Address{
+ Kind: HubStatus,
+ ID: parts[4],
+ }
case (len(parts) == 5 || len(parts) == 6) && parts[2] == "devices":
- address = Address{Kind: Kind(parts[4]), ID: parts[3]}
+ address = Address{
+ Kind: Kind(parts[4]),
+ ID: parts[3],
+ }
if len(parts) == 6 {
address.Target = parts[5]
if address.Kind == Command && address.Target == "$batch" {
// house-state input. Telemetry uses QoS 0. Other subscriptions use QoS 1.
func CoreSubscriptions() []Subscription {
return []Subscription{
- {Filter: Root + "/devices/+/descriptor", QoS: 1},
- {Filter: Root + "/devices/+/availability", QoS: 1},
- {Filter: Root + "/devices/+/reported/+", QoS: 1},
- {Filter: Root + "/devices/+/telemetry/+", QoS: 0},
- {Filter: Root + "/devices/+/events/+", QoS: 1},
- {Filter: Root + "/devices/+/results/+", QoS: 1},
+ {
+ Filter: Root + "/devices/+/descriptor",
+ QoS: 1,
+ },
+ {
+ Filter: Root + "/devices/+/availability",
+ QoS: 1,
+ },
+ {
+ Filter: Root + "/devices/+/reported/+",
+ QoS: 1,
+ },
+ {
+ Filter: Root + "/devices/+/telemetry/+",
+ QoS: 0,
+ },
+ {
+ Filter: Root + "/devices/+/events/+",
+ QoS: 1,
+ },
+ {
+ Filter: Root + "/devices/+/results/+",
+ QoS: 1,
+ },
}
}
return Subscription{}, fmt.Errorf("device ID must be a canonical ULID")
}
- return Subscription{Filter: Root + "/devices/" + deviceID + "/commands/+", QoS: 1}, nil
+ return Subscription{
+ Filter: Root + "/devices/" + deviceID + "/commands/+",
+ QoS: 1,
+ }, nil
}
// HubStatusSubscription returns the filter for all Zigbee hub diagnostics.
func HubStatusSubscription() Subscription {
- return Subscription{Filter: Root + "/zigbee/hubs/+/status", QoS: 1}
+ return Subscription{
+ Filter: Root + "/zigbee/hubs/+/status",
+ QoS: 1,
+ }
}
target string
suffix string
}{
- {name: "routes metadata", kind: Descriptor, suffix: "descriptor"},
- {name: "routes reachability", kind: Availability, suffix: "availability"},
- {name: "routes retained values", kind: Reported, target: "relay.1", suffix: "reported/relay.1"},
- {name: "routes samples", kind: Telemetry, target: "power", suffix: "telemetry/power"},
- {name: "routes occurrences", kind: Event, target: "press", suffix: "events/press"},
- {name: "routes requests", kind: Command, target: "relay", suffix: "commands/relay"},
- {name: "routes ordered requests", kind: CommandBatch, suffix: "commands/$batch"},
- {name: "routes outcomes", kind: Result, target: commandID, suffix: "results/" + commandID},
- {name: "routes hub diagnostics", kind: HubStatus, suffix: "status"},
+ {
+ name: "routes metadata",
+ kind: Descriptor,
+ suffix: "descriptor",
+ },
+ {
+ name: "routes reachability",
+ kind: Availability,
+ suffix: "availability",
+ },
+ {
+ name: "routes retained values",
+ kind: Reported,
+ target: "relay.1",
+ suffix: "reported/relay.1",
+ },
+ {
+ name: "routes samples",
+ kind: Telemetry,
+ target: "power",
+ suffix: "telemetry/power",
+ },
+ {
+ name: "routes occurrences",
+ kind: Event,
+ target: "press",
+ suffix: "events/press",
+ },
+ {
+ name: "routes requests",
+ kind: Command,
+ target: "relay",
+ suffix: "commands/relay",
+ },
+ {
+ name: "routes ordered requests",
+ kind: CommandBatch,
+ suffix: "commands/$batch",
+ },
+ {
+ name: "routes outcomes",
+ kind: Result,
+ target: commandID,
+ suffix: "results/" + commandID,
+ },
+ {
+ name: "routes hub diagnostics",
+ kind: HubStatus,
+ suffix: "status",
+ },
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
- address := Address{Kind: tc.kind, ID: deviceID, Target: tc.target}
+ address := Address{
+ Kind: tc.kind,
+ ID: deviceID,
+ Target: tc.target,
+ }
topic, err := address.Topic()
require.NoError(t, err)
func TestSubscriptionsSelectOnlyTheirDirection(t *testing.T) {
filters := CoreSubscriptions()
require.Len(t, filters, 6)
- require.Equal(t, Subscription{Filter: Root + "/devices/+/telemetry/+", QoS: 0}, filters[3])
+ require.Equal(t, Subscription{
+ Filter: Root + "/devices/+/telemetry/+",
+ QoS: 0,
+ }, filters[3])
for index, filter := range filters {
require.NotContains(t, filter.Filter, "commands")
--- /dev/null
+package protocol
+
+import (
+ "encoding/json"
+ "fmt"
+)
+
+// MaximumCommandBatchSize bounds the ordered commands in one publication.
+const MaximumCommandBatchSize = 256
+
+// BatchPayload holds 2 through 256 commands for one device. IDs and targets are
+// distinct. Array order is operation order. Each member keeps its own expiry,
+// correlation, and causation. There is no combined result.
+type BatchPayload struct {
+ Commands []BatchCommand
+}
+
+// MarshalJSON inserts each command's schema without caller-built envelopes.
+func (b BatchPayload) MarshalJSON() ([]byte, error) {
+ commands := make([]batchWireCommand, 0, len(b.Commands))
+
+ for _, command := range b.Commands {
+ commands = append(commands, batchWireCommand{
+ Schema: Schema,
+
+ Metadata: command.Metadata,
+
+ Capability: command.Capability,
+
+ Payload: command.Payload,
+ })
+ }
+
+ data, err := json.Marshal(batchWirePayload{Commands: commands})
+ if err != nil {
+ return nil, fmt.Errorf("encode batch members: %w", err)
+ }
+
+ return data, nil
+}
+
+// UnmarshalJSON rejects unknown fields and unsupported member schemas.
+func (b *BatchPayload) UnmarshalJSON(data []byte) error {
+ var wire batchWirePayload
+
+ err := decodeObject(data, &wire)
+ if err != nil {
+ return fmt.Errorf("decode batch members: %w", err)
+ }
+
+ commands := make([]BatchCommand, 0, len(wire.Commands))
+
+ for _, command := range wire.Commands {
+ if command.Schema != Schema {
+ return fmt.Errorf("unsupported batch member schema %d", command.Schema)
+ }
+
+ commands = append(commands, BatchCommand{
+ Metadata: command.Metadata,
+
+ Capability: command.Capability,
+
+ Payload: command.Payload,
+ })
+ }
+
+ b.Commands = commands
+
+ return nil
+}
+
+// BatchCommand holds one complete request without a topic or schema field. The
+// outer batch message ID equals the first member's message ID.
+type BatchCommand struct {
+ Metadata
+ Capability string
+ Payload ValuePayload
+}
+
+type batchWirePayload struct {
+ Commands []batchWireCommand `json:"commands"`
+}
+
+type batchWireCommand struct {
+ Schema int `json:"schema"`
+ Metadata
+ Capability string `json:"capability"`
+ Payload ValuePayload `json:"payload"`
+}
--- /dev/null
+package protocol
+
+import (
+ "encoding/json"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestBatchKeepsMemberMetadataAndEarliestDeadline(t *testing.T) {
+ deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ later := deadline.Add(time.Minute)
+ payload := testBatch(deadline)
+ payload.Commands[0].ExpiresAt = &later
+ payload.Commands[0].CorrelationID = deviceID
+ payload.Commands[1].CausationID = messageID
+ publication, err := BuildCommandBatch(deviceID, Metadata{MessageID: messageID}, payload)
+ require.NoError(t, err)
+ require.Equal(t, &deadline, publication.ExpiresAt)
+
+ parsed, err := ParseCommandBatch(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.Equal(t, payload, parsed.Payload)
+
+ for index := 2; index < MaximumCommandBatchSize; index++ {
+ metadata := Metadata{
+ MessageID: fmt.Sprintf("%026d", index+20),
+ ExpiresAt: &deadline,
+ }
+ payload.Commands = append(payload.Commands, BatchCommand{
+ Metadata: metadata,
+ Capability: fmt.Sprintf("relay.%d", index+1),
+ Payload: ValuePayload{Value: json.RawMessage(`false`)},
+ })
+ }
+
+ _, err = BuildCommandBatch(deviceID, Metadata{MessageID: messageID}, payload)
+ require.NoError(t, err)
+
+ payload.Commands = append(payload.Commands, payload.Commands[0])
+ _, err = BuildCommandBatch(deviceID, Metadata{MessageID: messageID}, payload)
+ require.Error(t, err)
+}
+
+func TestBatchRejectsMalformedMembersBeforeReturningCommands(t *testing.T) {
+ deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ metadata := Metadata{MessageID: messageID}
+ publication, err := BuildCommandBatch(deviceID, metadata, testBatch(deadline))
+ require.NoError(t, err)
+
+ testCases := []struct {
+ name string
+ old string
+ replacement string
+ }{
+ {
+ name: "rejects member schema",
+ old: `"schema":1`,
+ replacement: `"schema":2`,
+ },
+ {
+ name: "rejects unknown batch field",
+ old: `"commands":[`,
+ replacement: `"extra":1,"commands":[`,
+ },
+ {
+ name: "rejects unknown member field",
+ old: `"capability":"relay.1"`,
+ replacement: `"capability":"relay.1","extra":1`,
+ },
+ {
+ name: "rejects unknown member value field",
+ old: `"value":true`,
+ replacement: `"value":true,"extra":1`,
+ },
+ {
+ name: "rejects duplicate member ID",
+ old: commandID,
+ replacement: messageID,
+ },
+ {
+ name: "rejects duplicate target",
+ old: `"relay.2"`,
+ replacement: `"relay.1"`,
+ },
+ {
+ name: "rejects invalid target",
+ old: `"relay.2"`,
+ replacement: `"relay/+"`,
+ },
+ {
+ name: "requires member expiry",
+ old: `"expires_at":"2026-01-01T00:00:00Z",`,
+ replacement: ``,
+ },
+ {
+ name: "requires member value",
+ old: `"value":true`,
+ replacement: ``,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ data := strings.ReplaceAll(string(publication.Data), tc.old, tc.replacement)
+
+ var parsed Message[BatchPayload]
+
+ parsed, err = ParseCommandBatch(publication.Topic, []byte(data), false)
+ require.Error(t, err)
+ require.Empty(t, parsed.Payload.Commands)
+ })
+ }
+
+ payload := testBatch(deadline)
+ payload.Commands = payload.Commands[:1]
+ _, err = BuildCommandBatch(deviceID, metadata, payload)
+ require.Error(t, err)
+
+ _, err = BuildCommandBatch(deviceID, Metadata{MessageID: commandID}, testBatch(deadline))
+ require.Error(t, err)
+
+ data := strings.Replace(string(publication.Data), `"capability":"relay.1"`, `"capability":"relay.1","schema":2`, 1)
+ _, err = ParseCommandBatch(publication.Topic, []byte(data), false)
+ require.Error(t, err)
+}
--- /dev/null
+package protocol
+
+import "fmt"
+
+// BuildDescriptor returns a validated descriptor publication with its delivery
+// rules.
+func BuildDescriptor(deviceID string, metadata Metadata, payload DescriptorPayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: Descriptor,
+ ID: deviceID,
+ }, metadata, payload)
+}
+
+// ParseDescriptor validates its topic, envelope, payload, and delivery
+// retention.
+func ParseDescriptor(topic string, data []byte, retained bool) (Message[DescriptorPayload], error) {
+ message, err := parseMessage[DescriptorPayload](Descriptor, topic, data, retained)
+ if err != nil {
+ return Message[DescriptorPayload]{}, fmt.Errorf("parse descriptor: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildAvailability returns a validated availability publication with its
+// delivery rules.
+func BuildAvailability(deviceID string, metadata Metadata, payload AvailabilityPayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: Availability,
+ ID: deviceID,
+ }, metadata, payload)
+}
+
+// ParseAvailability validates its topic, envelope, payload, and delivery
+// retention.
+func ParseAvailability(topic string, data []byte, retained bool) (Message[AvailabilityPayload], error) {
+ message, err := parseMessage[AvailabilityPayload](Availability, topic, data, retained)
+ if err != nil {
+ return Message[AvailabilityPayload]{}, fmt.Errorf("parse availability: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildReported returns a validated reported publication with its delivery
+// rules.
+func BuildReported(deviceID string, capability string, metadata Metadata, payload ValuePayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: Reported,
+ ID: deviceID,
+ Target: capability,
+ }, metadata, payload)
+}
+
+// ParseReported validates its topic, envelope, payload, and delivery retention.
+func ParseReported(topic string, data []byte, retained bool) (Message[ValuePayload], error) {
+ message, err := parseMessage[ValuePayload](Reported, topic, data, retained)
+ if err != nil {
+ return Message[ValuePayload]{}, fmt.Errorf("parse reported: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildTelemetry returns a validated telemetry publication with its delivery
+// rules.
+func BuildTelemetry(deviceID string, capability string, metadata Metadata, payload ValuePayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: Telemetry,
+ ID: deviceID,
+ Target: capability,
+ }, metadata, payload)
+}
+
+// ParseTelemetry validates its topic, envelope, payload, and delivery
+// retention.
+func ParseTelemetry(topic string, data []byte, retained bool) (Message[ValuePayload], error) {
+ message, err := parseMessage[ValuePayload](Telemetry, topic, data, retained)
+ if err != nil {
+ return Message[ValuePayload]{}, fmt.Errorf("parse telemetry: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildEvent returns a validated event publication with its delivery rules.
+func BuildEvent(deviceID string, event string, metadata Metadata, payload EventPayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: Event,
+ ID: deviceID,
+ Target: event,
+ }, metadata, payload)
+}
+
+// ParseEvent validates its topic, envelope, payload, and delivery retention.
+func ParseEvent(topic string, data []byte, retained bool) (Message[EventPayload], error) {
+ message, err := parseMessage[EventPayload](Event, topic, data, retained)
+ if err != nil {
+ return Message[EventPayload]{}, fmt.Errorf("parse event: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildCommand returns a validated command publication with its delivery rules.
+func BuildCommand(deviceID string, capability string, metadata Metadata, payload ValuePayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: Command,
+ ID: deviceID,
+ Target: capability,
+ }, metadata, payload)
+}
+
+// ParseCommand validates its topic, envelope, payload, and delivery retention.
+func ParseCommand(topic string, data []byte, retained bool) (Message[ValuePayload], error) {
+ message, err := parseMessage[ValuePayload](Command, topic, data, retained)
+ if err != nil {
+ return Message[ValuePayload]{}, fmt.Errorf("parse command: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildCommandBatch returns a validated commandbatch publication with its
+// delivery rules.
+func BuildCommandBatch(deviceID string, metadata Metadata, payload BatchPayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: CommandBatch,
+ ID: deviceID,
+ }, metadata, payload)
+}
+
+// ParseCommandBatch validates its topic, envelope, payload, and delivery
+// retention.
+func ParseCommandBatch(topic string, data []byte, retained bool) (Message[BatchPayload], error) {
+ message, err := parseMessage[BatchPayload](CommandBatch, topic, data, retained)
+ if err != nil {
+ return Message[BatchPayload]{}, fmt.Errorf("parse commandbatch: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildResult returns a validated result publication with its delivery rules.
+func BuildResult(deviceID string, commandID string, metadata Metadata, payload ResultPayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: Result,
+ ID: deviceID,
+ Target: commandID,
+ }, metadata, payload)
+}
+
+// ParseResult validates its topic, envelope, payload, and delivery retention.
+func ParseResult(topic string, data []byte, retained bool) (Message[ResultPayload], error) {
+ message, err := parseMessage[ResultPayload](Result, topic, data, retained)
+ if err != nil {
+ return Message[ResultPayload]{}, fmt.Errorf("parse result: %w", err)
+ }
+
+ return message, nil
+}
+
+// BuildHubStatus returns a validated hubstatus publication with its delivery
+// rules.
+func BuildHubStatus(hubID string, metadata Metadata, payload HubStatusPayload) (Publication, error) {
+ return buildMessage(Address{
+ Kind: HubStatus,
+ ID: hubID,
+ }, metadata, payload)
+}
+
+// ParseHubStatus validates its topic, envelope, payload, and delivery
+// retention.
+func ParseHubStatus(topic string, data []byte, retained bool) (Message[HubStatusPayload], error) {
+ message, err := parseMessage[HubStatusPayload](HubStatus, topic, data, retained)
+ if err != nil {
+ return Message[HubStatusPayload]{}, fmt.Errorf("parse hubstatus: %w", err)
+ }
+
+ return message, nil
+}
--- /dev/null
+package protocol
+
+import (
+ "encoding/json"
+ "time"
+)
+
+func testDescriptor() DescriptorPayload {
+ return DescriptorPayload{
+ Manufacturer: "example",
+ Model: "relay",
+ Firmware: "1",
+ Attribution: "source",
+ Capabilities: []Capability{{
+ ID: "relay",
+ Type: "switch",
+ ReportsState: true,
+ AcceptsCommands: true,
+ }},
+ }
+}
+
+func testBatch(deadline time.Time) BatchPayload {
+ return BatchPayload{Commands: []BatchCommand{
+ {
+ Metadata: Metadata{
+ MessageID: messageID,
+ ExpiresAt: &deadline,
+ },
+ Capability: "relay.1",
+ Payload: ValuePayload{Value: json.RawMessage(`true`)},
+ },
+ {
+ Metadata: Metadata{
+ MessageID: commandID,
+ ExpiresAt: &deadline,
+ },
+ Capability: "relay.2",
+ Payload: ValuePayload{Value: json.RawMessage(`false`)},
+ },
+ }}
+}
--- /dev/null
+package protocol
+
+import (
+ "encoding/json"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestTypedPublicationsRoundTrip(t *testing.T) {
+ now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ deadline := now.Add(time.Minute)
+ metadata := Metadata{
+ MessageID: messageID,
+ SentAt: &now,
+ }
+ commandMetadata := Metadata{
+ MessageID: messageID,
+ ExpiresAt: &deadline,
+ }
+ resultMetadata := Metadata{
+ MessageID: messageID,
+ CorrelationID: commandID,
+ }
+ testCases := []struct {
+ name string
+ kind Kind
+ build func() (Publication, error)
+ payload any
+ retain bool
+ qos byte
+ }{
+ {
+ name: "preserves metadata contract",
+ kind: Descriptor,
+ retain: true,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildDescriptor(deviceID, metadata, testDescriptor())
+ },
+ payload: testDescriptor(),
+ },
+ {
+ name: "preserves online reachability",
+ kind: Availability,
+ retain: true,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildAvailability(deviceID, metadata, AvailabilityPayload{Status: "online"})
+ },
+ payload: AvailabilityPayload{Status: "online"},
+ },
+ {
+ name: "preserves retained object value",
+ kind: Reported,
+ retain: true,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildReported(deviceID, "weather", metadata, ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)})
+ },
+ payload: ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)},
+ },
+ {
+ name: "preserves unconfirmed sample",
+ kind: Telemetry,
+ qos: 0,
+
+ build: func() (Publication, error) {
+ return BuildTelemetry(deviceID, "power", metadata, ValuePayload{Value: json.RawMessage(`12.5`)})
+ },
+ payload: ValuePayload{Value: json.RawMessage(`12.5`)},
+ },
+ {
+ name: "preserves open event data",
+ kind: Event,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildEvent(deviceID, "motion", metadata, EventPayload{
+ "action": json.RawMessage(`"motion"`),
+ "custom": json.RawMessage(`7`),
+ })
+ },
+ payload: EventPayload{
+ "action": json.RawMessage(`"motion"`),
+ "custom": json.RawMessage(`7`),
+ },
+ },
+ {
+ name: "preserves command deadline",
+ kind: Command,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildCommand(deviceID, "relay", commandMetadata, ValuePayload{Value: json.RawMessage(`true`)})
+ },
+ payload: ValuePayload{Value: json.RawMessage(`true`)},
+ },
+ {
+ name: "preserves ordered member requests",
+ kind: CommandBatch,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildCommandBatch(deviceID, metadata, testBatch(deadline))
+ },
+ payload: testBatch(deadline),
+ },
+ {
+ name: "preserves correlated result",
+ kind: Result,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildResult(deviceID, commandID, resultMetadata, ResultPayload{Status: "completed"})
+ },
+ payload: ResultPayload{Status: "completed"},
+ },
+ {
+ name: "preserves expiring hub snapshot",
+ kind: HubStatus,
+ retain: true,
+ qos: 1,
+
+ build: func() (Publication, error) {
+ return BuildHubStatus(deviceID, commandMetadata, HubStatusPayload{
+ ID: deviceID,
+ Address: "example.com",
+ Status: "ready",
+ })
+ },
+ payload: HubStatusPayload{
+ ID: deviceID,
+ Address: "example.com",
+ Status: "ready",
+ },
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ publication, err := tc.build()
+ require.NoError(t, err)
+ require.Equal(t, tc.retain, publication.Retain)
+ require.Equal(t, tc.qos, publication.QoS)
+
+ parsed, err := Parse(publication.Topic, publication.Data, publication.Retain)
+ require.NoError(t, err)
+ require.Equal(t, tc.kind, parsed.Address.Kind)
+ require.Equal(t, deviceID, parsed.Address.ID)
+ require.Equal(t, tc.payload, parsed.Payload)
+ require.Equal(t, messageID, parsed.Metadata.MessageID)
+
+ live, err := Parse(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.False(t, live.Retained)
+
+ if !tc.retain {
+ _, err = Parse(publication.Topic, publication.Data, true)
+ require.Error(t, err)
+ }
+
+ malformed := strings.Replace(publication.Topic, Root, "wrong/v1", 1)
+ _, err = Parse(malformed, publication.Data, false)
+ require.Error(t, err)
+
+ unknown := strings.Replace(string(publication.Data), `"schema":1`, `"schema":1,"unknown":true`, 1)
+ _, err = Parse(publication.Topic, []byte(unknown), false)
+ require.Error(t, err)
+
+ wrongSchema := strings.Replace(string(publication.Data), `"schema":1`, `"schema":2`, 1)
+ _, err = Parse(publication.Topic, []byte(wrongSchema), false)
+ require.Error(t, err)
+
+ if tc.kind == Command || tc.kind == CommandBatch || tc.kind == HubStatus {
+ require.Equal(t, &deadline, publication.ExpiresAt)
+ }
+
+ if tc.kind == Command {
+ response, err := ParseTopic(publication.ResponseTopic)
+ require.NoError(t, err)
+ require.Equal(t, Result, response.Kind)
+ require.Equal(t, messageID, response.Target)
+ require.Equal(t, messageID, publication.CorrelationData)
+ } else {
+ require.Empty(t, publication.ResponseTopic)
+ require.Empty(t, publication.CorrelationData)
+ }
+ })
+ }
+}
+
+func TestTypedParsersRejectWrongOperations(t *testing.T) {
+ parsers := []func(string, []byte, bool) error{
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseDescriptor(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseAvailability(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseReported(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseTelemetry(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseEvent(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseCommand(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseCommandBatch(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseResult(topic, data, retained)
+ return err
+ },
+ func(topic string, data []byte, retained bool) error {
+ _, err := ParseHubStatus(topic, data, retained)
+ return err
+ },
+ }
+
+ for _, parse := range parsers {
+ require.Error(t, parse(Root+"/devices/"+deviceID+"/unknown", []byte(`{}`), false))
+ }
+}
--- /dev/null
+package protocol
+
+// DescriptorPayload describes machine metadata, not house names or rooms.
+// Attribution is optional and has a maximum length of 2000 bytes.
+type DescriptorPayload struct {
+ Manufacturer string `json:"manufacturer"`
+ Model string `json:"model"`
+ Firmware string `json:"firmware"`
+ Attribution string `json:"attribution,omitempty"`
+ Capabilities []Capability `json:"capabilities"`
+}
+
+// Capability describes one point. At least one access flag must be true. ID is
+// unique in its descriptor. ID and Type use protocol identifiers.
+type Capability struct {
+ ID string `json:"id"`
+ Type string `json:"type"`
+ ReportsState bool `json:"reports_state"`
+ AcceptsCommands bool `json:"accepts_commands"`
+}
// MeasurementRepeatInterval bounds equal retained measurement repeats.
MeasurementRepeatInterval = 5 * time.Minute
+
// OnlineRepeatInterval bounds equal online availability repeats.
OnlineRepeatInterval = 5 * time.Minute
+
// DeviceContactTimeout bounds the age of device contact evidence.
DeviceContactTimeout = 15 * time.Minute
)
return nil, fmt.Errorf("encode payload: %w", err)
}
- envelope := Envelope{Schema: Schema, Metadata: metadata, Payload: encoded}
+ envelope := Envelope{
+ Schema: Schema,
+ Metadata: metadata,
+ Payload: encoded,
+ }
err = validateEnvelope(envelope)
if err != nil {
return nil, fmt.Errorf("validate envelope: %w", err)
func TestEnvelopeRoundTripPreservesMetadata(t *testing.T) {
now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
- metadata := Metadata{MessageID: messageID, SentAt: &now, ExpiresAt: &now, CorrelationID: commandID, CausationID: deviceID}
+ metadata := Metadata{
+ MessageID: messageID,
+ SentAt: &now,
+ ExpiresAt: &now,
+ CorrelationID: commandID,
+ CausationID: deviceID,
+ }
data, err := EncodeEnvelope(metadata, map[string]bool{"value": true})
require.NoError(t, err)
name string
data string
}{
- {name: "rejects unsupported schema", data: strings.Replace(valid, `"schema":1`, `"schema":2`, 1)},
- {name: "rejects unknown fields", data: strings.Replace(valid, `"payload":{}`, `"payload":{},"extra":true`, 1)},
- {name: "rejects missing payload", data: `{"schema":1,"message_id":"` + messageID + `"}`},
- {name: "rejects null payload", data: strings.Replace(valid, `"payload":{}`, `"payload":null`, 1)},
- {name: "rejects non-object payload", data: strings.Replace(valid, `"payload":{}`, `"payload":[]`, 1)},
- {name: "rejects missing identity", data: `{"schema":1,"payload":{}}`},
- {name: "rejects invalid correlation", data: strings.Replace(valid, `"payload":{}`, `"payload":{},"correlation_id":"bad"`, 1)},
- {name: "rejects invalid causation", data: strings.Replace(valid, `"payload":{}`, `"payload":{},"causation_id":"bad"`, 1)},
- {name: "rejects non-UTC time", data: strings.Replace(valid, `"payload":{}`, `"payload":{},"sent_at":"2026-01-01T01:00:00+01:00"`, 1)},
- {name: "rejects invalid expiry", data: strings.Replace(valid, `"payload":{}`, `"payload":{},"expires_at":"bad"`, 1)},
- {name: "rejects trailing JSON", data: valid + `{}`},
+ {
+ name: "rejects unsupported schema",
+ data: strings.Replace(valid, `"schema":1`, `"schema":2`, 1),
+ },
+ {
+ name: "rejects unknown fields",
+ data: strings.Replace(valid, `"payload":{}`, `"payload":{},"extra":true`, 1),
+ },
+ {
+ name: "rejects missing payload",
+ data: `{"schema":1,"message_id":"` + messageID + `"}`,
+ },
+ {
+ name: "rejects null payload",
+ data: strings.Replace(valid, `"payload":{}`, `"payload":null`, 1),
+ },
+ {
+ name: "rejects non-object payload",
+ data: strings.Replace(valid, `"payload":{}`, `"payload":[]`, 1),
+ },
+ {
+ name: "rejects missing identity",
+ data: `{"schema":1,"payload":{}}`,
+ },
+ {
+ name: "rejects invalid correlation",
+ data: strings.Replace(valid, `"payload":{}`, `"payload":{},"correlation_id":"bad"`, 1),
+ },
+ {
+ name: "rejects invalid causation",
+ data: strings.Replace(valid, `"payload":{}`, `"payload":{},"causation_id":"bad"`, 1),
+ },
+ {
+ name: "rejects non-UTC time",
+ data: strings.Replace(valid, `"payload":{}`, `"payload":{},"sent_at":"2026-01-01T01:00:00+01:00"`, 1),
+ },
+ {
+ name: "rejects invalid expiry",
+ data: strings.Replace(valid, `"payload":{}`, `"payload":{},"expires_at":"bad"`, 1),
+ },
+ {
+ name: "rejects trailing JSON",
+ data: valid + `{}`,
+ },
}
+
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
_, err := ParseEnvelope([]byte(tc.data))
--- /dev/null
+package protocol
+
+import (
+ "fmt"
+ "time"
+)
+
+// HubStatusPayload holds a hub snapshot, not house state. ID matches the hub
+// topic. Status is ready or unavailable. ExpiresAt is required in its envelope.
+// Address is runtime data and must not be written to logs.
+type HubStatusPayload struct {
+ ID string `json:"id"`
+ Alias string `json:"alias,omitempty"`
+ Address string `json:"address"`
+ Status string `json:"status"`
+ Error string `json:"error,omitempty"`
+ Model string `json:"model,omitempty"`
+ Firmware string `json:"firmware,omitempty"`
+ EventStream bool `json:"event_stream"`
+ UpdatedAt *time.Time `json:"updated_at,omitempty"`
+ Pairing *HubPairing `json:"pairing,omitempty"`
+ ConfiguredDevices int `json:"configured_devices"`
+ PairedDevices *int `json:"paired_devices,omitempty"`
+ Diagnostics *HubDiagnostics `json:"diagnostics,omitempty"`
+ DiagnosticsError string `json:"diagnostics_error,omitempty"`
+}
+
+// HubPairing describes one request to permit device joining.
+type HubPairing struct {
+ Status string `json:"status"`
+ DurationSeconds int `json:"duration_seconds"`
+ RequestedAt time.Time `json:"requested_at"`
+ Deadline time.Time `json:"deadline"`
+ Error string `json:"error,omitempty"`
+}
+
+// HubDiagnostics holds a source sample. Repeated publications preserve
+// ObservedAt and Stale. Optional measurements remain absent when unknown.
+type HubDiagnostics struct {
+ ObservedAt time.Time `json:"observed_at"`
+ Stale bool `json:"stale"`
+ Hostname string `json:"hostname,omitempty"`
+ IPAddress string `json:"ip_address,omitempty"`
+ MAC string `json:"mac,omitempty"`
+ RadioChip string `json:"radio_chip,omitempty"`
+ RadioFirmware string `json:"radio_firmware,omitempty"`
+ UptimeSeconds *uint64 `json:"uptime_seconds,omitempty"`
+ TemperatureCelsius *float64 `json:"temperature_celsius,omitempty"`
+ RadioTemperatureCelsius *float64 `json:"radio_temperature_celsius,omitempty"`
+ RAMUsedPercent *float64 `json:"ram_used_percent,omitempty"`
+ PSRAMUsedPercent *float64 `json:"psram_used_percent,omitempty"`
+ StorageUsedPercent *float64 `json:"storage_used_percent,omitempty"`
+ EthernetConnected *bool `json:"ethernet_connected,omitempty"`
+ WiFiConnected *bool `json:"wifi_connected,omitempty"`
+}
+
+// HubStatusLifetime bounds retained hub status to two polls plus 50 seconds. A
+// nonpositive poll interval is invalid.
+func HubStatusLifetime(pollInterval time.Duration) (time.Duration, error) {
+ if pollInterval <= 0 || pollInterval > (time.Duration(1<<63-1)-50*time.Second)/2 {
+ return 0, fmt.Errorf("hub poll interval is outside the supported range")
+ }
+
+ return 2*pollInterval + 50*time.Second, nil
+}
--- /dev/null
+package protocol
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestHubSnapshotPreservesDiagnosticsAndExpiry(t *testing.T) {
+ now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ lifetime, err := HubStatusLifetime(time.Minute)
+ require.NoError(t, err)
+ require.Equal(t, 170*time.Second, lifetime)
+
+ deadline := now.Add(lifetime)
+ count := 3
+ temperature := 24.5
+ payload := HubStatusPayload{
+ ID: deviceID,
+ Address: "example.com",
+ Alias: "hub",
+ Status: "ready",
+ EventStream: true,
+ UpdatedAt: &now,
+ ConfiguredDevices: 3,
+ PairedDevices: &count,
+
+ Pairing: &HubPairing{
+ Status: "accepted",
+ DurationSeconds: 60,
+ RequestedAt: now,
+ Deadline: now.Add(time.Minute),
+ },
+
+ Diagnostics: &HubDiagnostics{
+ ObservedAt: now,
+ Stale: true,
+ TemperatureCelsius: &temperature,
+ },
+ }
+ publication, err := BuildHubStatus(deviceID, Metadata{
+ MessageID: messageID,
+ ExpiresAt: &deadline,
+ }, payload)
+ require.NoError(t, err)
+
+ parsed, err := ParseHubStatus(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.Equal(t, payload, parsed.Payload)
+ require.Equal(t, &deadline, publication.ExpiresAt)
+
+ payload.Status = "unavailable"
+ _, err = BuildHubStatus(deviceID, Metadata{
+ MessageID: messageID,
+ ExpiresAt: &deadline,
+ }, payload)
+ require.NoError(t, err)
+
+ nonUTC := now.In(time.FixedZone("test", 3600))
+ payload.UpdatedAt = &nonUTC
+ _, err = BuildHubStatus(deviceID, Metadata{
+ MessageID: messageID,
+ ExpiresAt: &deadline,
+ }, payload)
+ require.Error(t, err)
+
+ for _, interval := range []time.Duration{0, -1, 1<<63 - 1} {
+ _, err = HubStatusLifetime(interval)
+ require.Error(t, err)
+ }
+}
--- /dev/null
+package protocol
+
+import (
+ "encoding/json"
+ "fmt"
+ "time"
+)
+
+// Message contains validated routing, metadata, and a typed payload. Retained
+// is the delivery flag. Live delivery of retained-topic data can be false.
+type Message[T any] struct {
+ Address Address
+ Metadata Metadata
+ Payload T
+ Retained bool
+}
+
+// Publication is a broker-independent outgoing message. The transport sets
+// UTF-8 payload format and application/json content type. Commands and hub
+// status also need MQTT expiry computed from their envelope deadlines.
+type Publication struct {
+ Topic string
+ Data []byte
+ QoS byte
+ Retain bool
+ // ExpiresAt is the command, earliest batch member, or hub deadline.
+ ExpiresAt *time.Time
+ // ResponseTopic and CorrelationData apply to individual commands only.
+ ResponseTopic string
+ CorrelationData string
+}
+
+// Parse routes an incoming publication to a typed payload without caller topic
+// operations. Payload is DescriptorPayload, AvailabilityPayload, ValuePayload,
+// EventPayload, BatchPayload, ResultPayload, or HubStatusPayload. Address.Kind
+// distinguishes command, reported, and telemetry value payloads.
+func Parse(topic string, data []byte, retained bool) (Message[any], error) {
+ address, err := ParseTopic(topic)
+ if err != nil {
+ return Message[any]{}, fmt.Errorf("parse address: %w", err)
+ }
+
+ switch address.Kind {
+ case Descriptor:
+ return eraseMessage(ParseDescriptor(topic, data, retained))
+ case Availability:
+ return eraseMessage(ParseAvailability(topic, data, retained))
+ case Reported:
+ return eraseMessage(ParseReported(topic, data, retained))
+ case Telemetry:
+ return eraseMessage(ParseTelemetry(topic, data, retained))
+ case Event:
+ return eraseMessage(ParseEvent(topic, data, retained))
+ case Command:
+ return eraseMessage(ParseCommand(topic, data, retained))
+ case CommandBatch:
+ return eraseMessage(ParseCommandBatch(topic, data, retained))
+ case Result:
+ return eraseMessage(ParseResult(topic, data, retained))
+ case HubStatus:
+ return eraseMessage(ParseHubStatus(topic, data, retained))
+ default:
+ return Message[any]{}, fmt.Errorf("unsupported message kind")
+ }
+}
+
+func eraseMessage[T any](message Message[T], err error) (Message[any], error) {
+ if err != nil {
+ return Message[any]{}, fmt.Errorf("parse message: %w", err)
+ }
+
+ return Message[any]{
+ Address: message.Address,
+ Metadata: message.Metadata,
+ Payload: message.Payload,
+ Retained: message.Retained,
+ }, nil
+}
+
+func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) {
+ topic, err := address.Topic()
+ if err != nil {
+ return Publication{}, fmt.Errorf("build address: %w", err)
+ }
+
+ data, err := EncodeEnvelope(metadata, payload)
+ if err != nil {
+ return Publication{}, fmt.Errorf("build envelope: %w", err)
+ }
+
+ retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
+ _, err = parseMessage[T](address.Kind, topic, data, retain)
+ if err != nil {
+ return Publication{}, fmt.Errorf("validate publication: %w", err)
+ }
+
+ qos := byte(1)
+ if address.Kind == Telemetry {
+ qos = 0
+ }
+
+ publication := Publication{
+ Topic: topic,
+ Data: data,
+ QoS: qos,
+ Retain: retain,
+ }
+
+ switch address.Kind {
+ case Command:
+ publication.ExpiresAt = metadata.ExpiresAt
+ publication.CorrelationData = metadata.MessageID
+ publication.ResponseTopic, err = (Address{
+ Kind: Result,
+ ID: address.ID,
+ Target: metadata.MessageID,
+ }).Topic()
+ if err != nil {
+ return Publication{}, fmt.Errorf("build command response topic: %w", err)
+ }
+ case CommandBatch:
+ batch, ok := any(payload).(BatchPayload)
+ if !ok {
+ return Publication{}, fmt.Errorf("batch requires command members")
+ }
+
+ for _, command := range batch.Commands {
+ if publication.ExpiresAt == nil || command.ExpiresAt.Before(*publication.ExpiresAt) {
+ publication.ExpiresAt = command.ExpiresAt
+ }
+ }
+ case HubStatus:
+ publication.ExpiresAt = metadata.ExpiresAt
+ }
+
+ return publication, nil
+}
+
+func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (Message[T], error) {
+ address, err := ParseTopic(topic)
+ if err != nil {
+ return Message[T]{}, fmt.Errorf("parse address: %w", err)
+ }
+
+ if address.Kind != kind {
+ return Message[T]{}, fmt.Errorf("message topic does not match parser")
+ }
+
+ if retained && (kind == Telemetry || kind == Event || kind == Command || kind == CommandBatch || kind == Result) {
+ return Message[T]{}, fmt.Errorf("message must not be retained")
+ }
+
+ envelope, err := ParseEnvelope(data)
+ if err != nil {
+ return Message[T]{}, fmt.Errorf("parse envelope: %w", err)
+ }
+
+ var payload T
+
+ err = decodeObject(envelope.Payload, &payload)
+ if err != nil {
+ return Message[T]{}, fmt.Errorf("decode payload: %w", err)
+ }
+
+ err = validatePayload(address, envelope, payload)
+ if err != nil {
+ return Message[T]{}, fmt.Errorf("validate payload: %w", err)
+ }
+
+ return Message[T]{
+ Address: address,
+ Metadata: envelope.Metadata,
+ Payload: payload,
+ Retained: retained,
+ }, nil
+}
+
+func validatePayload(address Address, envelope Envelope, payload any) error {
+ switch value := payload.(type) {
+ case DescriptorPayload:
+ err := validateDescriptor(value, envelope.Payload)
+ if err != nil {
+ return fmt.Errorf("validate descriptor: %w", err)
+ }
+ case AvailabilityPayload:
+ if value.Status != "online" && value.Status != "offline" {
+ return fmt.Errorf("availability must be online or offline")
+ }
+ case ValuePayload:
+ if !json.Valid(value.Value) {
+ return fmt.Errorf("value is required and must be valid JSON")
+ }
+
+ if address.Kind == Command && envelope.ExpiresAt == nil {
+ return fmt.Errorf("command must have an expiry")
+ }
+ case EventPayload:
+ // An event object has application-defined fields.
+ case ResultPayload:
+ if envelope.CorrelationID != address.Target {
+ return fmt.Errorf("result correlation must match its command ID")
+ }
+
+ switch value.Status {
+ case "accepted", "completed":
+ if value.Reason != "" {
+ return fmt.Errorf("successful result must not have a reason")
+ }
+ case "rejected", "failed":
+ if value.Reason == "" {
+ return fmt.Errorf("unsuccessful result must have a reason")
+ }
+ default:
+ return fmt.Errorf("result status is invalid")
+ }
+ case BatchPayload:
+ err := validateBatch(value, envelope)
+ if err != nil {
+ return fmt.Errorf("validate command batch: %w", err)
+ }
+ case HubStatusPayload:
+ err := validateHub(value, address, envelope)
+ if err != nil {
+ return fmt.Errorf("validate hub status: %w", err)
+ }
+ default:
+ return fmt.Errorf("unsupported payload type")
+ }
+
+ return nil
+}
+
+func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
+ if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(payload.Attribution) > 2000 || len(payload.Capabilities) == 0 {
+ return fmt.Errorf("descriptor metadata or capabilities are invalid")
+ }
+
+ var fields struct {
+ Capabilities []map[string]json.RawMessage `json:"capabilities"`
+ }
+
+ err := json.Unmarshal(raw, &fields)
+ if err != nil {
+ return fmt.Errorf("read descriptor access flags: %w", err)
+ }
+
+ seen := make(map[string]bool, len(payload.Capabilities))
+
+ for index, capability := range payload.Capabilities {
+ if !ValidIdentifier(capability.ID) || !ValidIdentifier(capability.Type) || seen[capability.ID] {
+ return fmt.Errorf("capability IDs and types must be valid and unique")
+ }
+
+ if !capability.ReportsState && !capability.AcceptsCommands {
+ return fmt.Errorf("capability must have at least one access mode")
+ }
+
+ for _, name := range []string{"reports_state", "accepts_commands"} {
+ field := fields.Capabilities[index][name]
+ if string(field) != "true" && string(field) != "false" {
+ return fmt.Errorf("capability access flags must be present booleans")
+ }
+ }
+
+ seen[capability.ID] = true
+ }
+
+ return nil
+}
+
+func validateBatch(payload BatchPayload, envelope Envelope) error {
+ if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
+ return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
+ }
+
+ if envelope.MessageID != payload.Commands[0].MessageID {
+ return fmt.Errorf("batch ID must equal its first command ID")
+ }
+
+ ids := make(map[string]bool, len(payload.Commands))
+ targets := make(map[string]bool, len(payload.Commands))
+
+ for _, command := range payload.Commands {
+ if !ValidIdentifier(command.Capability) || ids[command.MessageID] || targets[command.Capability] {
+ return fmt.Errorf("batch targets and IDs must be valid and distinct")
+ }
+
+ _, err := EncodeEnvelope(command.Metadata, command.Payload)
+ if err != nil {
+ return fmt.Errorf("validate batch command envelope: %w", err)
+ }
+
+ if command.ExpiresAt == nil || !json.Valid(command.Payload.Value) {
+ return fmt.Errorf("batch commands require expiry and a JSON value")
+ }
+
+ ids[command.MessageID] = true
+ targets[command.Capability] = true
+ }
+
+ return nil
+}
+
+func validateHub(payload HubStatusPayload, address Address, envelope Envelope) error {
+ if payload.ID != address.ID || envelope.ExpiresAt == nil {
+ return fmt.Errorf("hub status requires a matching ID and an expiry")
+ }
+
+ if payload.Status != "ready" && payload.Status != "unavailable" {
+ return fmt.Errorf("hub status must be ready or unavailable")
+ }
+
+ if payload.Address == "" || payload.ConfiguredDevices < 0 || payload.PairedDevices != nil && *payload.PairedDevices < 0 {
+ return fmt.Errorf("hub address or device counts are invalid")
+ }
+
+ timestamps := []*time.Time{payload.UpdatedAt}
+ if payload.Diagnostics != nil {
+ timestamps = append(timestamps, &payload.Diagnostics.ObservedAt)
+ }
+
+ if payload.Pairing != nil {
+ timestamps = append(timestamps, &payload.Pairing.RequestedAt, &payload.Pairing.Deadline)
+ }
+
+ for _, timestamp := range timestamps {
+ if timestamp != nil {
+ _, offset := timestamp.Zone()
+ if offset != 0 || timestamp.IsZero() {
+ return fmt.Errorf("hub timestamps must be nonzero UTC times")
+ }
+ }
+ }
+
+ return nil
+}
--- /dev/null
+package protocol
+
+import (
+ "encoding/json"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestPayloadValidationRejectsInvalidContracts(t *testing.T) {
+ deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ metadata := Metadata{
+ MessageID: messageID,
+ ExpiresAt: &deadline,
+ }
+ descriptor := `{"manufacturer":"example","model":"relay","firmware":"1","capabilities":[{"id":"relay","type":"switch","reports_state":true,"accepts_commands":false}]}`
+ testCases := []struct {
+ name string
+ kind Kind
+ target string
+ payload string
+ }{
+ {
+ name: "rejects unknown descriptor field",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `"model":"relay"`, `"model":"relay","room":"kitchen"`, 1),
+ },
+ {
+ name: "rejects unknown capability field",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `"type":"switch"`, `"type":"switch","extra":true`, 1),
+ },
+ {
+ name: "requires both access flags",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `,"accepts_commands":false`, ``, 1),
+ },
+ {
+ name: "rejects null access flag",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `"accepts_commands":false`, `"accepts_commands":null`, 1),
+ },
+ {
+ name: "requires active access mode",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `"reports_state":true`, `"reports_state":false`, 1),
+ },
+ {
+ name: "requires device metadata",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `"model":"relay"`, `"model":""`, 1),
+ },
+ {
+ name: "requires capabilities",
+ kind: Descriptor,
+ payload: `{"manufacturer":"example","model":"relay","firmware":"1","capabilities":[]}`,
+ },
+ {
+ name: "rejects invalid capability identifier",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `"id":"relay"`, `"id":"relay/1"`, 1),
+ },
+ {
+ name: "rejects duplicate capability",
+ kind: Descriptor,
+ payload: strings.Replace(descriptor, `}]`, `},{"id":"relay","type":"switch","reports_state":true,"accepts_commands":false}]`, 1),
+ },
+ {
+ name: "rejects unknown availability status",
+ kind: Availability,
+ payload: `{"status":"ready"}`,
+ },
+ {
+ name: "rejects unknown availability field",
+ kind: Availability,
+ payload: `{"status":"online","extra":1}`,
+ },
+ {
+ name: "requires report value",
+ kind: Reported,
+ target: "relay",
+ payload: `{}`,
+ },
+ {
+ name: "rejects unknown report field",
+ kind: Reported,
+ target: "relay",
+ payload: `{"value":true,"extra":1}`,
+ },
+ {
+ name: "requires telemetry value",
+ kind: Telemetry,
+ target: "power",
+ payload: `{}`,
+ },
+ {
+ name: "rejects unknown telemetry field",
+ kind: Telemetry,
+ target: "power",
+ payload: `{"value":1,"extra":1}`,
+ },
+ {
+ name: "rejects unknown command field",
+ kind: Command,
+ target: "relay",
+ payload: `{"value":true,"extra":1}`,
+ },
+ {
+ name: "requires command value",
+ kind: Command,
+ target: "relay",
+ payload: `{}`,
+ },
+ {
+ name: "rejects array event",
+ kind: Event,
+ target: "press",
+ payload: `[]`,
+ },
+ {
+ name: "rejects unknown result field",
+ kind: Result,
+ target: commandID,
+ payload: `{"status":"completed","extra":1}`,
+ },
+ {
+ name: "rejects unknown hub field",
+ kind: HubStatus,
+ payload: `{"id":"` + deviceID + `","status":"ready","address":"example.com","extra":1}`,
+ },
+ {
+ name: "rejects unknown diagnostic field",
+ kind: HubStatus,
+ payload: `{"id":"` + deviceID + `","status":"ready","address":"example.com","diagnostics":{"observed_at":"2026-01-01T00:00:00Z","stale":false,"extra":1}}`,
+ },
+ {
+ name: "rejects unknown pairing field",
+ kind: HubStatus,
+ payload: `{"id":"` + deviceID + `","status":"ready","address":"example.com","pairing":{"extra":1}}`,
+ },
+ {
+ name: "rejects hub identity mismatch",
+ kind: HubStatus,
+ payload: `{"id":"` + commandID + `","status":"ready","address":"example.com"}`,
+ },
+ {
+ name: "rejects invalid hub status",
+ kind: HubStatus,
+ payload: `{"id":"` + deviceID + `","status":"online","address":"example.com"}`,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ data, err := EncodeEnvelope(metadata, json.RawMessage(tc.payload))
+ if tc.kind == Event {
+ require.Error(t, err)
+
+ return
+ }
+
+ require.NoError(t, err)
+
+ topic, err := (Address{
+ Kind: tc.kind,
+ ID: deviceID,
+ Target: tc.target,
+ }).Topic()
+ require.NoError(t, err)
+
+ _, err = Parse(topic, data, false)
+ require.Error(t, err)
+ })
+ }
+}
+
+func TestBuildersApplyCommandAndResultRules(t *testing.T) {
+ metadata := Metadata{MessageID: messageID}
+ _, err := BuildCommand(deviceID, "relay", metadata, ValuePayload{Value: json.RawMessage(`true`)})
+ require.Error(t, err)
+
+ _, err = BuildHubStatus(deviceID, metadata, HubStatusPayload{
+ ID: deviceID,
+ Address: "example.com",
+ Status: "ready",
+ })
+ require.Error(t, err)
+
+ _, err = BuildDescriptor("bad", metadata, testDescriptor())
+ require.Error(t, err)
+
+ _, err = BuildReported(deviceID, "relay", metadata, ValuePayload{Value: json.RawMessage(`bad`)})
+ require.Error(t, err)
+
+ _, err = BuildAvailability(deviceID, metadata, AvailabilityPayload{Status: "bad"})
+ require.Error(t, err)
+
+ for _, value := range []string{`null`, `true`, `12`, `"open"`, `{"mode":"off"}`, `[1,2]`} {
+ var publication Publication
+
+ publication, err = BuildReported(deviceID, "point", metadata, ValuePayload{Value: json.RawMessage(value)})
+ require.NoError(t, err)
+
+ var message Message[ValuePayload]
+
+ message, err = ParseReported(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.JSONEq(t, value, string(message.Payload.Value))
+ }
+
+ testCases := []struct {
+ name string
+ status string
+ reason string
+ valid bool
+ }{
+ {
+ name: "accepts intermediate result",
+ status: "accepted",
+ valid: true,
+ },
+ {
+ name: "accepts completed result",
+ status: "completed",
+ valid: true,
+ },
+ {
+ name: "accepts rejection reason",
+ status: "rejected",
+ reason: "unsupported",
+ valid: true,
+ },
+ {
+ name: "accepts failure reason",
+ status: "failed",
+ reason: "readback",
+ valid: true,
+ },
+ {
+ name: "requires rejection reason",
+ status: "rejected",
+ },
+ {
+ name: "requires failure reason",
+ status: "failed",
+ },
+ {
+ name: "rejects successful reason",
+ status: "completed",
+ reason: "extra",
+ },
+ {
+ name: "rejects unknown result status",
+ status: "expired",
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ correlated := Metadata{
+ MessageID: messageID,
+ CorrelationID: commandID,
+ }
+
+ var publication Publication
+
+ publication, err = BuildResult(deviceID, commandID, correlated, ResultPayload{
+ Status: tc.status,
+ Reason: tc.reason,
+ })
+ if !tc.valid {
+ require.Error(t, err)
+
+ return
+ }
+
+ require.NoError(t, err)
+
+ var parsed Message[ResultPayload]
+
+ parsed, err = ParseResult(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.Equal(t, tc.status, parsed.Payload.Status)
+ })
+ }
+
+ _, err = BuildResult(deviceID, commandID, metadata, ResultPayload{Status: "completed"})
+ require.Error(t, err)
+
+ descriptor := testDescriptor()
+ descriptor.Attribution = strings.Repeat("x", 2001)
+ _, err = BuildDescriptor(deviceID, metadata, descriptor)
+ require.Error(t, err)
+}
--- /dev/null
+package protocol
+
+import "encoding/json"
+
+// ValuePayload holds a complete value for reports, telemetry, or commands.
+// Value is required. JSON null is a present value. Capability-specific value
+// validation belongs to the driver or entity, not to the wire layer.
+type ValuePayload struct {
+ Value json.RawMessage `json:"value"`
+}
+
+// AvailabilityPayload holds reachability. Status is online or offline.
+type AvailabilityPayload struct {
+ Status string `json:"status"`
+}
+
+// EventPayload is an application-defined occurrence object. Every key is event
+// data, not an optional wire field. Empty objects are valid. Button and camera
+// actions can use action and occurred_at without a closed field list.
+type EventPayload map[string]json.RawMessage
+
+// ResultPayload holds accepted, completed, rejected, or failed. Only rejected
+// and failed require a nonempty Reason. Other statuses must omit Reason.
+type ResultPayload struct {
+ Status string `json:"status"`
+ Reason string `json:"reason,omitempty"`
+}
projects / onehouse / commitdiff