+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
+}
projects / onehouse / commitdiff