X-Git-Url: https://git.giacomolli.uk/onehouse/blobdiff_plain/4bba0e163f1a4dc07b34fd9e4dc403de1f9b4cb9..c87607884883cca4e47736db79a11c03f175138f:/protocol/message.go?ds=inline

diff --git a/protocol/message.go b/protocol/message.go
index 0fd4ec9..50a2e1c 100644
--- a/protocol/message.go
+++ b/protocol/message.go
@@ -42,58 +42,47 @@ func Parse(topic string, data []byte, retained bool) (Message[any], error) {
 
 	switch address.Kind {
 	case Descriptor:
-		return eraseMessage(ParseDescriptor(topic, data, retained))
+		return eraseMessage(parsePayload[DescriptorPayload](address, data, retained))
 	case Availability:
-		return eraseMessage(ParseAvailability(topic, data, retained))
+		return eraseMessage(parsePayload[AvailabilityPayload](address, data, retained))
 	case Reported:
-		return eraseMessage(ParseReported(topic, data, retained))
+		return eraseMessage(parsePayload[ValuePayload](address, data, retained))
 	case Telemetry:
-		return eraseMessage(ParseTelemetry(topic, data, retained))
+		return eraseMessage(parsePayload[ValuePayload](address, data, retained))
 	case Event:
-		return eraseMessage(ParseEvent(topic, data, retained))
+		return eraseMessage(parsePayload[EventPayload](address, data, retained))
 	case Command:
-		return eraseMessage(ParseCommand(topic, data, retained))
+		return eraseMessage(parsePayload[ValuePayload](address, data, retained))
 	case CommandBatch:
-		return eraseMessage(ParseCommandBatch(topic, data, retained))
+		return eraseMessage(parsePayload[BatchPayload](address, data, retained))
 	case Result:
-		return eraseMessage(ParseResult(topic, data, retained))
+		return eraseMessage(parsePayload[ResultPayload](address, data, retained))
 	case HubStatus:
-		return eraseMessage(ParseHubStatus(topic, data, retained))
+		return eraseMessage(parsePayload[HubStatusPayload](address, 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)
+	normalized := normalizePayload(payload)
+	err = validatePayload(address, metadata, normalized)
 	if err != nil {
-		return Publication{}, fmt.Errorf("build envelope: %w", err)
+		return Publication{}, fmt.Errorf("validate publication: %w", err)
 	}
 
-	retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
-	_, err = parseMessage[T](address.Kind, topic, data, retain)
+	data, err := encodeEnvelope(metadata, normalized)
 	if err != nil {
-		return Publication{}, fmt.Errorf("validate publication: %w", err)
+		return Publication{}, fmt.Errorf("build envelope: %w", err)
 	}
 
+	retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
+
 	qos := byte(1)
 	if address.Kind == Telemetry {
 		qos = 0
@@ -146,6 +135,30 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
 		return Message[T]{}, fmt.Errorf("message topic does not match parser")
 	}
 
+	message, err := parsePayload[T](address, data, retained)
+	if err != nil {
+		return Message[T]{}, fmt.Errorf("parse payload: %w", err)
+	}
+
+	return message, nil
+}
+
+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 parsePayload[T any](address Address, data []byte, retained bool) (Message[T], error) {
+	kind := address.Kind
+
 	if retained && (kind == Telemetry || kind == Event || kind == Command || kind == CommandBatch || kind == Result) {
 		return Message[T]{}, fmt.Errorf("message must not be retained")
 	}
@@ -162,11 +175,18 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
 		return Message[T]{}, fmt.Errorf("decode payload: %w", err)
 	}
 
-	err = validatePayload(address, envelope, payload)
+	err = validatePayload(address, envelope.Metadata, payload)
 	if err != nil {
 		return Message[T]{}, fmt.Errorf("validate payload: %w", err)
 	}
 
+	if address.Kind == Descriptor {
+		err = validateDescriptorFlags(envelope.Payload)
+		if err != nil {
+			return Message[T]{}, fmt.Errorf("validate descriptor access flags: %w", err)
+		}
+	}
+
 	return Message[T]{
 		Address:  address,
 		Metadata: envelope.Metadata,
@@ -175,66 +195,64 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M
 	}, nil
 }
 
-func validatePayload(address Address, envelope wireEnvelope, payload any) error {
+// normalizePayload keeps nil raw values as JSON null without changing caller
+// data.
+func normalizePayload(payload any) any {
 	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 value.Value == nil {
+			value.Value = json.RawMessage(`null`)
 		}
 
-		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")
-		}
+		return value
+	case BatchPayload:
+		commands := make([]BatchCommand, len(value.Commands))
+		copy(commands, value.Commands)
 
-		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")
+		for index := range commands {
+			if commands[index].Payload.Value == nil {
+				commands[index].Payload.Value = json.RawMessage(`null`)
 			}
-		default:
-			return fmt.Errorf("result status is invalid")
 		}
+
+		value.Commands = commands
+
+		return value
+	default:
+		return payload
+	}
+}
+
+func validatePayload(address Address, metadata Metadata, payload any) error {
+	var err error
+
+	switch value := payload.(type) {
+	case DescriptorPayload:
+		err = validateDescriptor(value)
+	case AvailabilityPayload:
+		err = validateAvailability(value)
+	case ValuePayload:
+		err = validateValue(value, address.Kind, metadata)
+	case EventPayload:
+		err = validateEvent(value)
+	case ResultPayload:
+		err = validateResult(value, address, metadata)
 	case BatchPayload:
-		err := validateBatch(value, envelope)
-		if err != nil {
-			return fmt.Errorf("validate command batch: %w", err)
-		}
+		err = validateBatch(value, metadata)
 	case HubStatusPayload:
-		err := validateHub(value, address, envelope)
-		if err != nil {
-			return fmt.Errorf("validate hub status: %w", err)
-		}
+		err = validateHub(value, address, metadata)
 	default:
 		return fmt.Errorf("unsupported payload type")
 	}
 
+	if err != nil {
+		return fmt.Errorf("validate %s payload: %w", address.Kind, err)
+	}
+
 	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")
-	}
-
+func validateDescriptorFlags(raw json.RawMessage) error {
 	var fields struct {
 		Capabilities []map[string]json.RawMessage `json:"capabilities"`
 	}
@@ -244,9 +262,26 @@ func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
 		return fmt.Errorf("read descriptor access flags: %w", err)
 	}
 
+	for _, capability := range fields.Capabilities {
+		for _, name := range []string{"reports_state", "accepts_commands"} {
+			field := capability[name]
+			if string(field) != "true" && string(field) != "false" {
+				return fmt.Errorf("capability access flags must be present booleans")
+			}
+		}
+	}
+
+	return nil
+}
+
+func validateDescriptor(payload DescriptorPayload) error {
+	if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(string([]rune(payload.Attribution))) > 2000 || len(payload.Capabilities) == 0 {
+		return fmt.Errorf("descriptor metadata or capabilities are invalid")
+	}
+
 	seen := make(map[string]bool, len(payload.Capabilities))
 
-	for index, capability := range payload.Capabilities {
+	for _, 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")
 		}
@@ -255,25 +290,73 @@ func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
 			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 validateAvailability(payload AvailabilityPayload) error {
+	if payload.Status != "online" && payload.Status != "offline" {
+		return fmt.Errorf("availability must be online or offline")
+	}
+
+	return nil
+}
+
+func validateValue(payload ValuePayload, kind Kind, metadata Metadata) error {
+	if !json.Valid(payload.Value) {
+		return fmt.Errorf("value is required and must be valid JSON")
+	}
+
+	if kind == Command && metadata.ExpiresAt == nil {
+		return fmt.Errorf("command must have an expiry")
+	}
+
+	return nil
+}
+
+func validateEvent(payload EventPayload) error {
+	if payload == nil {
+		return fmt.Errorf("event must be a JSON object")
+	}
+
+	for _, value := range payload {
+		if value != nil && !json.Valid(value) {
+			return fmt.Errorf("event values must be valid JSON")
 		}
+	}
 
-		seen[capability.ID] = true
+	return nil
+}
+
+func validateResult(payload ResultPayload, address Address, metadata Metadata) error {
+	if metadata.CorrelationID != address.Target {
+		return fmt.Errorf("result correlation must match its command ID")
+	}
+
+	switch payload.Status {
+	case "accepted", "completed":
+		if payload.Reason != "" {
+			return fmt.Errorf("successful result must not have a reason")
+		}
+	case "rejected", "failed":
+		if payload.Reason == "" {
+			return fmt.Errorf("unsuccessful result must have a reason")
+		}
+	default:
+		return fmt.Errorf("result status is invalid")
 	}
 
 	return nil
 }
 
-func validateBatch(payload BatchPayload, envelope wireEnvelope) error {
+func validateBatch(payload BatchPayload, metadata Metadata) 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 {
+	if metadata.MessageID != payload.Commands[0].MessageID {
 		return fmt.Errorf("batch ID must equal its first command ID")
 	}
 
@@ -285,13 +368,14 @@ func validateBatch(payload BatchPayload, envelope wireEnvelope) error {
 			return fmt.Errorf("batch targets and IDs must be valid and distinct")
 		}
 
-		_, err := encodeEnvelope(command.Metadata, command.Payload)
+		err := validateMetadata(command.Metadata)
 		if err != nil {
-			return fmt.Errorf("validate batch command envelope: %w", err)
+			return fmt.Errorf("validate batch command metadata: %w", err)
 		}
 
-		if command.ExpiresAt == nil || !json.Valid(command.Payload.Value) {
-			return fmt.Errorf("batch commands require expiry and a JSON value")
+		err = validateValue(command.Payload, Command, command.Metadata)
+		if err != nil {
+			return fmt.Errorf("validate batch command value: %w", err)
 		}
 
 		ids[command.MessageID] = true
@@ -301,8 +385,8 @@ func validateBatch(payload BatchPayload, envelope wireEnvelope) error {
 	return nil
 }
 
-func validateHub(payload HubStatusPayload, address Address, envelope wireEnvelope) error {
-	if payload.ID != address.ID || envelope.ExpiresAt == nil {
+func validateHub(payload HubStatusPayload, address Address, metadata Metadata) error {
+	if payload.ID != address.ID || metadata.ExpiresAt == nil {
 		return fmt.Errorf("hub status requires a matching ID and an expiry")
 	}