From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 21:42:10 +0000 (-0300) Subject: Keep topic and envelope construction private X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/4bba0e163f1a4dc07b34fd9e4dc403de1f9b4cb9?hp=b28ae53ae04c41116acb2beae9d5a2128a187c83 Keep topic and envelope construction private Typed publications and subscription helpers own the wire layout. The public API exposes parsed addresses but no low-level topic or envelope construction, which the driver framework does not need. Timing constants have their own file, and batch fields and operation names use the normal layout and words without changing the wire contract. Issue: 8fa2f25 --- diff --git a/doc/mqtt.rst b/doc/mqtt.rst index 55a9a67..1f5fa4a 100644 --- a/doc/mqtt.rst +++ b/doc/mqtt.rst @@ -7,17 +7,17 @@ command, and message IDs use canonical uppercase ULIDs. Capabilities and events use lowercase ASCII letters and digits, with single internal hyphen or period separators. Topic builders reject invalid addresses. -``Address.Topic`` and ``ParseTopic`` cover device descriptor, -availability, reported values, telemetry, events, commands, command -batches, results, and Zigbee hub status. Subscription helpers supply the -core input filters, each device's command filter, and hub status filter. -The command filter includes the reserved ``$batch`` target. - -``EncodeEnvelope`` and ``ParseEnvelope`` validate schema, IDs, optional -UTC timestamps, and a required object payload. Unknown envelope fields, -trailing JSON, unsupported schema versions, and malformed topics are -errors. A receiver checks expiry against its own clock. The message ID -stays unchanged on retry. Its timestamp is not the observation time. +Private topic helpers cover device descriptor, availability, reported +values, telemetry, events, commands, command batches, results, and +Zigbee hub status. Subscription helpers supply the core input filters, +each device's command filter, and hub status filter. The command filter +includes the reserved ``$batch`` target. + +Private envelope helpers validate schema, IDs, optional UTC timestamps, +and a required object payload. Unknown envelope fields, trailing JSON, +unsupported schema versions, and malformed topics are errors. A receiver +checks expiry against its own clock. The message ID 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 diff --git a/protocol/address.go b/protocol/address.go index fbd0d26..7786dfd 100644 --- a/protocol/address.go +++ b/protocol/address.go @@ -48,8 +48,8 @@ type Address struct { Target string } -// Topic validates the address and returns its exact publication topic. -func (a Address) Topic() (string, error) { +// topic validates the address and returns its exact publication topic. +func (a Address) topic() (string, error) { if !ValidID(a.ID) { return "", fmt.Errorf("address ID must be a canonical ULID") } @@ -88,8 +88,8 @@ func (a Address) Topic() (string, error) { return Root + "/devices/" + a.ID + "/" + string(a.Kind) + "/" + a.Target, nil } -// ParseTopic rejects malformed topics, wildcards, and unsupported operations. -func ParseTopic(topic string) (Address, error) { +// parseTopic rejects malformed topics, wildcards, and unsupported operations. +func parseTopic(topic string) (Address, error) { parts := strings.Split(topic, "/") if len(parts) < 5 || parts[0]+"/"+parts[1] != Root { return Address{}, fmt.Errorf("topic must use the versioned root") @@ -119,7 +119,7 @@ func ParseTopic(topic string) (Address, error) { return Address{}, fmt.Errorf("topic layout is invalid") } - canonical, err := address.Topic() + canonical, err := address.topic() if err != nil { return Address{}, fmt.Errorf("validate topic address: %w", err) } diff --git a/protocol/address_test.go b/protocol/address_test.go index eca76d7..6325b62 100644 --- a/protocol/address_test.go +++ b/protocol/address_test.go @@ -79,7 +79,7 @@ func TestAddressRoundTrip(t *testing.T) { ID: deviceID, Target: tc.target, } - topic, err := address.Topic() + topic, err := address.topic() require.NoError(t, err) expected := Root + "/devices/" + deviceID + "/" + tc.suffix @@ -89,7 +89,7 @@ func TestAddressRoundTrip(t *testing.T) { require.Equal(t, expected, topic) - parsed, err := ParseTopic(topic) + parsed, err := parseTopic(topic) require.NoError(t, err) require.Equal(t, address, parsed) }) @@ -113,7 +113,7 @@ func TestParseTopicRejectsMalformedAddresses(t *testing.T) { } for _, tc := range testCases { - _, err := ParseTopic(tc) + _, err := parseTopic(tc) require.Error(t, err, tc) } } diff --git a/protocol/batch_payload.go b/protocol/batch_payload.go index 3670939..6e0f07b 100644 --- a/protocol/batch_payload.go +++ b/protocol/batch_payload.go @@ -21,13 +21,10 @@ func (b BatchPayload) MarshalJSON() ([]byte, error) { for _, command := range b.Commands { commands = append(commands, batchWireCommand{ - Schema: Schema, - - Metadata: command.Metadata, - + Schema: Schema, + Metadata: command.Metadata, Capability: command.Capability, - - Payload: command.Payload, + Payload: command.Payload, }) } @@ -56,11 +53,9 @@ func (b *BatchPayload) UnmarshalJSON(data []byte) error { } commands = append(commands, BatchCommand{ - Metadata: command.Metadata, - + Metadata: command.Metadata, Capability: command.Capability, - - Payload: command.Payload, + Payload: command.Payload, }) } diff --git a/protocol/builders.go b/protocol/builders.go index b9a39f7..16f94f8 100644 --- a/protocol/builders.go +++ b/protocol/builders.go @@ -121,7 +121,7 @@ func ParseCommand(topic string, data []byte, retained bool) (Message[ValuePayloa return message, nil } -// BuildCommandBatch returns a validated commandbatch publication with its +// BuildCommandBatch returns a validated command batch publication with its // delivery rules. func BuildCommandBatch(deviceID string, metadata Metadata, payload BatchPayload) (Publication, error) { return buildMessage(Address{ @@ -135,7 +135,7 @@ func BuildCommandBatch(deviceID string, metadata Metadata, payload BatchPayload) 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[BatchPayload]{}, fmt.Errorf("parse command batch: %w", err) } return message, nil @@ -160,7 +160,7 @@ func ParseResult(topic string, data []byte, retained bool) (Message[ResultPayloa return message, nil } -// BuildHubStatus returns a validated hubstatus publication with its delivery +// BuildHubStatus returns a validated hub status publication with its delivery // rules. func BuildHubStatus(hubID string, metadata Metadata, payload HubStatusPayload) (Publication, error) { return buildMessage(Address{ @@ -174,7 +174,7 @@ func BuildHubStatus(hubID string, metadata Metadata, payload HubStatusPayload) ( 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[HubStatusPayload]{}, fmt.Errorf("parse hub status: %w", err) } return message, nil diff --git a/protocol/builders_test.go b/protocol/builders_test.go index 159cca2..b2b5a4a 100644 --- a/protocol/builders_test.go +++ b/protocol/builders_test.go @@ -182,7 +182,7 @@ func TestTypedPublicationsRoundTrip(t *testing.T) { } if tc.kind == Command { - response, err := ParseTopic(publication.ResponseTopic) + response, err := parseTopic(publication.ResponseTopic) require.NoError(t, err) require.Equal(t, Result, response.Kind) require.Equal(t, messageID, response.Target) diff --git a/protocol/message.go b/protocol/message.go index 94ea7b5..0fd4ec9 100644 --- a/protocol/message.go +++ b/protocol/message.go @@ -35,7 +35,7 @@ type Publication struct { // 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) + address, err := parseTopic(topic) if err != nil { return Message[any]{}, fmt.Errorf("parse address: %w", err) } @@ -78,12 +78,12 @@ func eraseMessage[T any](message Message[T], err error) (Message[any], error) { } func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) { - topic, err := address.Topic() + topic, err := address.topic() if err != nil { return Publication{}, fmt.Errorf("build address: %w", err) } - data, err := EncodeEnvelope(metadata, payload) + data, err := encodeEnvelope(metadata, payload) if err != nil { return Publication{}, fmt.Errorf("build envelope: %w", err) } @@ -114,7 +114,7 @@ func buildMessage[T any](address Address, metadata Metadata, payload T) (Publica Kind: Result, ID: address.ID, Target: metadata.MessageID, - }).Topic() + }).topic() if err != nil { return Publication{}, fmt.Errorf("build command response topic: %w", err) } @@ -137,7 +137,7 @@ func buildMessage[T any](address Address, metadata Metadata, payload T) (Publica } func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (Message[T], error) { - address, err := ParseTopic(topic) + address, err := parseTopic(topic) if err != nil { return Message[T]{}, fmt.Errorf("parse address: %w", err) } @@ -150,7 +150,7 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M return Message[T]{}, fmt.Errorf("message must not be retained") } - envelope, err := ParseEnvelope(data) + envelope, err := parseEnvelope(data) if err != nil { return Message[T]{}, fmt.Errorf("parse envelope: %w", err) } @@ -175,7 +175,7 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M }, nil } -func validatePayload(address Address, envelope Envelope, payload any) error { +func validatePayload(address Address, envelope wireEnvelope, payload any) error { switch value := payload.(type) { case DescriptorPayload: err := validateDescriptor(value, envelope.Payload) @@ -268,7 +268,7 @@ func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error { return nil } -func validateBatch(payload BatchPayload, envelope Envelope) error { +func validateBatch(payload BatchPayload, envelope wireEnvelope) error { if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize { return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize) } @@ -285,7 +285,7 @@ func validateBatch(payload BatchPayload, envelope Envelope) error { return fmt.Errorf("batch targets and IDs must be valid and distinct") } - _, err := EncodeEnvelope(command.Metadata, command.Payload) + _, err := encodeEnvelope(command.Metadata, command.Payload) if err != nil { return fmt.Errorf("validate batch command envelope: %w", err) } @@ -301,7 +301,7 @@ func validateBatch(payload BatchPayload, envelope Envelope) error { return nil } -func validateHub(payload HubStatusPayload, address Address, envelope Envelope) error { +func validateHub(payload HubStatusPayload, address Address, envelope wireEnvelope) error { if payload.ID != address.ID || envelope.ExpiresAt == nil { return fmt.Errorf("hub status requires a matching ID and an expiry") } diff --git a/protocol/message_test.go b/protocol/message_test.go index 770ae50..79871dd 100644 --- a/protocol/message_test.go +++ b/protocol/message_test.go @@ -154,7 +154,7 @@ func TestPayloadValidationRejectsInvalidContracts(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - data, err := EncodeEnvelope(metadata, json.RawMessage(tc.payload)) + data, err := encodeEnvelope(metadata, json.RawMessage(tc.payload)) if tc.kind == Event { require.Error(t, err) @@ -167,7 +167,7 @@ func TestPayloadValidationRejectsInvalidContracts(t *testing.T) { Kind: tc.kind, ID: deviceID, Target: tc.target, - }).Topic() + }).topic() require.NoError(t, err) _, err = Parse(topic, data, false) diff --git a/protocol/timing.go b/protocol/timing.go new file mode 100644 index 0000000..e7fd410 --- /dev/null +++ b/protocol/timing.go @@ -0,0 +1,14 @@ +package protocol + +import "time" + +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 +) diff --git a/protocol/envelope.go b/protocol/wire_envelope.go similarity index 70% rename from protocol/envelope.go rename to protocol/wire_envelope.go index a6833a4..5e8b778 100644 --- a/protocol/envelope.go +++ b/protocol/wire_envelope.go @@ -9,19 +9,8 @@ import ( "time" ) -const ( - // Schema is the supported envelope and payload schema version. - Schema = 1 - - // 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 -) +// Schema is the supported envelope and payload schema version. +const Schema = 1 // Metadata holds message identity and optional UTC times. Retries keep the same // ID. A receiver checks ExpiresAt against its clock before processing. @@ -33,21 +22,21 @@ type Metadata struct { CausationID string `json:"causation_id,omitempty"` } -// Envelope carries one JSON object and its schema and metadata. -type Envelope struct { +// wireEnvelope carries one JSON object and its schema and metadata. +type wireEnvelope struct { Schema int `json:"schema"` Metadata Payload json.RawMessage `json:"payload"` } -// EncodeEnvelope validates metadata and encodes a schema-1 object payload. -func EncodeEnvelope(metadata Metadata, payload any) ([]byte, error) { +// encodeEnvelope validates metadata and encodes a schema-1 object payload. +func encodeEnvelope(metadata Metadata, payload any) ([]byte, error) { encoded, err := json.Marshal(payload) if err != nil { return nil, fmt.Errorf("encode payload: %w", err) } - envelope := Envelope{ + envelope := wireEnvelope{ Schema: Schema, Metadata: metadata, Payload: encoded, @@ -65,25 +54,25 @@ func EncodeEnvelope(metadata Metadata, payload any) ([]byte, error) { return encoded, nil } -// ParseEnvelope rejects unknown fields, trailing JSON, unsupported schemas, +// parseEnvelope rejects unknown fields, trailing JSON, unsupported schemas, // invalid IDs, non-UTC times, and missing or non-object payloads. -func ParseEnvelope(data []byte) (Envelope, error) { - var envelope Envelope +func parseEnvelope(data []byte) (wireEnvelope, error) { + var parsed wireEnvelope - err := decodeObject(data, &envelope) + err := decodeObject(data, &parsed) if err != nil { - return Envelope{}, fmt.Errorf("decode envelope: %w", err) + return wireEnvelope{}, fmt.Errorf("decode envelope: %w", err) } - err = validateEnvelope(envelope) + err = validateEnvelope(parsed) if err != nil { - return Envelope{}, fmt.Errorf("validate envelope: %w", err) + return wireEnvelope{}, fmt.Errorf("validate envelope: %w", err) } - return envelope, nil + return parsed, nil } -func validateEnvelope(envelope Envelope) error { +func validateEnvelope(envelope wireEnvelope) error { if envelope.Schema != Schema { return fmt.Errorf("unsupported schema %d", envelope.Schema) } diff --git a/protocol/envelope_test.go b/protocol/wire_envelope_test.go similarity index 87% rename from protocol/envelope_test.go rename to protocol/wire_envelope_test.go index 4b55451..ce11f91 100644 --- a/protocol/envelope_test.go +++ b/protocol/wire_envelope_test.go @@ -17,10 +17,10 @@ func TestEnvelopeRoundTripPreservesMetadata(t *testing.T) { CorrelationID: commandID, CausationID: deviceID, } - data, err := EncodeEnvelope(metadata, map[string]bool{"value": true}) + data, err := encodeEnvelope(metadata, map[string]bool{"value": true}) require.NoError(t, err) - envelope, err := ParseEnvelope(data) + envelope, err := parseEnvelope(data) require.NoError(t, err) require.Equal(t, Schema, envelope.Schema) require.Equal(t, metadata, envelope.Metadata) @@ -85,17 +85,17 @@ func TestEnvelopeRejectsInvalidWireData(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - _, err := ParseEnvelope([]byte(tc.data)) + _, err := parseEnvelope([]byte(tc.data)) require.Error(t, err) }) } - _, err := EncodeEnvelope(Metadata{MessageID: "bad"}, map[string]bool{}) + _, err := encodeEnvelope(Metadata{MessageID: "bad"}, map[string]bool{}) require.Error(t, err) - _, err = EncodeEnvelope(Metadata{MessageID: messageID}, make(chan int)) + _, err = encodeEnvelope(Metadata{MessageID: messageID}, make(chan int)) require.Error(t, err) - _, err = EncodeEnvelope(Metadata{MessageID: messageID}, []int{1}) + _, err = encodeEnvelope(Metadata{MessageID: messageID}, []int{1}) require.Error(t, err) }