From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 20:52:01 +0000 (-0300) Subject: Build typed MQTT publications X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/b28ae53ae04c41116acb2beae9d5a2128a187c83?ds=sidebyside Build typed MQTT publications Typed builders and parsers cover every wire operation, including ordered command batches and expiring hub diagnostics. Publications carry their delivery rules and command response metadata, so callers need no topic or envelope construction. Strict closed payloads reject unsupported fields while event objects keep complete application data. Issue: 8fa2f25 --- diff --git a/doc/mqtt.rst b/doc/mqtt.rst index 615116e..55a9a67 100644 --- a/doc/mqtt.rst +++ b/doc/mqtt.rst @@ -22,3 +22,112 @@ stays unchanged on retry. Its timestamp is not the observation time. 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. diff --git a/protocol/address.go b/protocol/address.go index 31d1b45..fbd0d26 100644 --- a/protocol/address.go +++ b/protocol/address.go @@ -11,20 +11,28 @@ const ( // 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" ) @@ -91,9 +99,15 @@ func ParseTopic(topic string) (Address, error) { 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" { @@ -127,12 +141,30 @@ type Subscription struct { // 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, + }, } } @@ -142,10 +174,16 @@ func CommandSubscription(deviceID string) (Subscription, error) { 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, + } } diff --git a/protocol/address_test.go b/protocol/address_test.go index f6600c8..eca76d7 100644 --- a/protocol/address_test.go +++ b/protocol/address_test.go @@ -20,20 +20,65 @@ func TestAddressRoundTrip(t *testing.T) { 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) @@ -76,7 +121,10 @@ func TestParseTopicRejectsMalformedAddresses(t *testing.T) { 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") diff --git a/protocol/batch_payload.go b/protocol/batch_payload.go new file mode 100644 index 0000000..3670939 --- /dev/null +++ b/protocol/batch_payload.go @@ -0,0 +1,89 @@ +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"` +} diff --git a/protocol/batch_payload_test.go b/protocol/batch_payload_test.go new file mode 100644 index 0000000..4f268b3 --- /dev/null +++ b/protocol/batch_payload_test.go @@ -0,0 +1,129 @@ +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) +} diff --git a/protocol/builders.go b/protocol/builders.go new file mode 100644 index 0000000..b9a39f7 --- /dev/null +++ b/protocol/builders.go @@ -0,0 +1,181 @@ +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 +} diff --git a/protocol/builders_helper_test.go b/protocol/builders_helper_test.go new file mode 100644 index 0000000..67662a3 --- /dev/null +++ b/protocol/builders_helper_test.go @@ -0,0 +1,42 @@ +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`)}, + }, + }} +} diff --git a/protocol/builders_test.go b/protocol/builders_test.go new file mode 100644 index 0000000..159cca2 --- /dev/null +++ b/protocol/builders_test.go @@ -0,0 +1,241 @@ +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)) + } +} diff --git a/protocol/descriptor_payload.go b/protocol/descriptor_payload.go new file mode 100644 index 0000000..c02f943 --- /dev/null +++ b/protocol/descriptor_payload.go @@ -0,0 +1,20 @@ +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"` +} diff --git a/protocol/envelope.go b/protocol/envelope.go index 9db8c03..a6833a4 100644 --- a/protocol/envelope.go +++ b/protocol/envelope.go @@ -15,8 +15,10 @@ const ( // 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 ) @@ -45,7 +47,11 @@ func EncodeEnvelope(metadata Metadata, payload any) ([]byte, error) { 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) diff --git a/protocol/envelope_test.go b/protocol/envelope_test.go index dfb0600..4b55451 100644 --- a/protocol/envelope_test.go +++ b/protocol/envelope_test.go @@ -10,7 +10,13 @@ import ( 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) @@ -31,18 +37,52 @@ func TestEnvelopeRejectsInvalidWireData(t *testing.T) { 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)) diff --git a/protocol/hub_status_payload.go b/protocol/hub_status_payload.go new file mode 100644 index 0000000..a8e6ff9 --- /dev/null +++ b/protocol/hub_status_payload.go @@ -0,0 +1,65 @@ +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 +} diff --git a/protocol/hub_status_payload_test.go b/protocol/hub_status_payload_test.go new file mode 100644 index 0000000..8d834fb --- /dev/null +++ b/protocol/hub_status_payload_test.go @@ -0,0 +1,72 @@ +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) + } +} diff --git a/protocol/message.go b/protocol/message.go new file mode 100644 index 0000000..94ea7b5 --- /dev/null +++ b/protocol/message.go @@ -0,0 +1,336 @@ +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 +} diff --git a/protocol/message_test.go b/protocol/message_test.go new file mode 100644 index 0000000..770ae50 --- /dev/null +++ b/protocol/message_test.go @@ -0,0 +1,296 @@ +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) +} diff --git a/protocol/value_payload.go b/protocol/value_payload.go new file mode 100644 index 0000000..4658e71 --- /dev/null +++ b/protocol/value_payload.go @@ -0,0 +1,27 @@ +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"` +}