From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 20:42:52 +0000 (-0300) Subject: Define wire addresses and envelopes X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/9ea825d64a3efaf8679990ea652ffaad0615c24c?ds=inline;hp=7ef0ef316c8765dc42f704c26bf57eba28808def Define wire addresses and envelopes The protocol package validates versioned topics, message metadata, and subscription filters. Canonical IDs and fixed publication intervals let callers share one wire contract without a broker dependency. Tests use testify, with its dependencies pinned in the Nix package. Issue: 8fa2f25 --- diff --git a/doc/mqtt.rst b/doc/mqtt.rst new file mode 100644 index 0000000..615116e --- /dev/null +++ b/doc/mqtt.rst @@ -0,0 +1,24 @@ +############### + MQTT contract +############### + +The ``protocol`` package defines schema 1 under ``onehouse/v1``. Device, +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. + +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. diff --git a/go.mod b/go.mod index 9e58e09..8444aae 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,11 @@ module onehouse go 1.26.5 + +require github.com/stretchr/testify v1.11.1 + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..c4c1710 --- /dev/null +++ b/go.sum @@ -0,0 +1,10 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/nix/package.nix b/nix/package.nix index 6b37eeb..cbf8459 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -9,7 +9,7 @@ buildGoModule { inherit src version; subPackages = [ "cmd/onehouse" ]; - vendorHash = null; + vendorHash = "sha256-yqtj+zCo7u2UwaQ12bHCHPNucgQNkvkN7nfkLynB67Y="; doInstallCheck = true; installCheckPhase = '' diff --git a/protocol/address.go b/protocol/address.go new file mode 100644 index 0000000..31d1b45 --- /dev/null +++ b/protocol/address.go @@ -0,0 +1,151 @@ +package protocol + +import ( + "fmt" + "strings" +) + +const ( + // Root is the versioned MQTT namespace. + Root = "onehouse/v1" + + // 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" +) + +// Kind identifies a wire operation, independent of its topic text. +type Kind string + +// Address identifies a device operation or a Zigbee hub status. Target is a +// capability, event, or command ID. Hub status and device metadata omit Target. +type Address struct { + Kind Kind + ID string + Target string +} + +// 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") + } + + switch a.Kind { + case Descriptor, Availability: + if a.Target != "" { + return "", fmt.Errorf("metadata address must not have a target") + } + + return Root + "/devices/" + a.ID + "/" + string(a.Kind), nil + case Reported, Telemetry, Event, Command: + if !ValidIdentifier(a.Target) { + return "", fmt.Errorf("point address must have a valid target") + } + case Result: + if !ValidID(a.Target) { + return "", fmt.Errorf("result target must be a canonical ULID") + } + case CommandBatch: + if a.Target != "" { + return "", fmt.Errorf("batch address must not have a target") + } + + return Root + "/devices/" + a.ID + "/commands/$batch", nil + case HubStatus: + if a.Target != "" { + return "", fmt.Errorf("hub address must not have a target") + } + + return Root + "/zigbee/hubs/" + a.ID + "/status", nil + default: + return "", fmt.Errorf("unknown address kind %q", a.Kind) + } + + 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) { + parts := strings.Split(topic, "/") + if len(parts) < 5 || parts[0]+"/"+parts[1] != Root { + return Address{}, fmt.Errorf("topic must use the versioned root") + } + + var address Address + + switch { + case len(parts) == 6 && parts[2] == "zigbee" && parts[3] == "hubs" && parts[5] == "status": + 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]} + if len(parts) == 6 { + address.Target = parts[5] + if address.Kind == Command && address.Target == "$batch" { + address.Kind = CommandBatch + address.Target = "" + } + } + default: + return Address{}, fmt.Errorf("topic layout is invalid") + } + + canonical, err := address.Topic() + if err != nil { + return Address{}, fmt.Errorf("validate topic address: %w", err) + } + + if topic != canonical { + return Address{}, fmt.Errorf("topic is not canonical") + } + + return address, nil +} + +// Subscription contains a topic filter and the requested delivery level. +type Subscription struct { + Filter string + QoS byte +} + +// CoreSubscriptions returns device-to-core filters. Hub diagnostics are not +// 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}, + } +} + +// CommandSubscription returns the device command filter, including batches. +func CommandSubscription(deviceID string) (Subscription, error) { + if !ValidID(deviceID) { + return Subscription{}, fmt.Errorf("device ID must be a canonical ULID") + } + + 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} +} diff --git a/protocol/address_test.go b/protocol/address_test.go new file mode 100644 index 0000000..f6600c8 --- /dev/null +++ b/protocol/address_test.go @@ -0,0 +1,97 @@ +package protocol + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +const ( + deviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAV" + messageID = "01ARZ3NDEKTSV4RRFFQ69G5FAW" + commandID = "01ARZ3NDEKTSV4RRFFQ69G5FAX" +) + +func TestAddressRoundTrip(t *testing.T) { + testCases := []struct { + name string + kind Kind + 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"}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + address := Address{Kind: tc.kind, ID: deviceID, Target: tc.target} + topic, err := address.Topic() + require.NoError(t, err) + + expected := Root + "/devices/" + deviceID + "/" + tc.suffix + if tc.kind == HubStatus { + expected = Root + "/zigbee/hubs/" + deviceID + "/status" + } + + require.Equal(t, expected, topic) + + parsed, err := ParseTopic(topic) + require.NoError(t, err) + require.Equal(t, address, parsed) + }) + } +} + +func TestParseTopicRejectsMalformedAddresses(t *testing.T) { + testCases := []string{ + "", "onehouse/v2/devices/" + deviceID + "/descriptor", + Root + "/devices/" + strings.ToLower(deviceID) + "/descriptor", + Root + "/devices/" + deviceID + "/descriptor/extra", + Root + "/devices/" + deviceID + "/reported", + Root + "/devices/" + deviceID + "/reported/", + Root + "/devices/" + deviceID + "/reported/+", + Root + "/devices/" + deviceID + "/events/door--open", + Root + "/devices/" + deviceID + "/commands/$other", + Root + "/devices/" + deviceID + "/results/not-an-id", + Root + "/devices/" + deviceID + "/status", + Root + "/zigbee/hubs/" + deviceID + "/status/extra", + Root + "/devices/" + deviceID + "/unknown", + } + + for _, tc := range testCases { + _, err := ParseTopic(tc) + require.Error(t, err, tc) + } +} + +func TestSubscriptionsSelectOnlyTheirDirection(t *testing.T) { + filters := CoreSubscriptions() + require.Len(t, filters, 6) + require.Equal(t, Subscription{Filter: Root + "/devices/+/telemetry/+", QoS: 0}, filters[3]) + + for index, filter := range filters { + require.NotContains(t, filter.Filter, "commands") + + if index != 3 { + require.Equal(t, byte(1), filter.QoS) + } + } + + subscription, err := CommandSubscription(deviceID) + require.NoError(t, err) + require.Equal(t, Root+"/devices/"+deviceID+"/commands/+", subscription.Filter) + require.Equal(t, byte(1), subscription.QoS) + + _, err = CommandSubscription("+") + require.Error(t, err) + require.Equal(t, Root+"/zigbee/hubs/+/status", HubStatusSubscription().Filter) +} diff --git a/protocol/envelope.go b/protocol/envelope.go new file mode 100644 index 0000000..9db8c03 --- /dev/null +++ b/protocol/envelope.go @@ -0,0 +1,139 @@ +package protocol + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "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 +) + +// Metadata holds message identity and optional UTC times. Retries keep the same +// ID. A receiver checks ExpiresAt against its clock before processing. +type Metadata struct { + MessageID string `json:"message_id"` + SentAt *time.Time `json:"sent_at,omitempty"` + ExpiresAt *time.Time `json:"expires_at,omitempty"` + CorrelationID string `json:"correlation_id,omitempty"` + CausationID string `json:"causation_id,omitempty"` +} + +// Envelope carries one JSON object and its schema and metadata. +type Envelope 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) { + encoded, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("encode payload: %w", err) + } + + envelope := Envelope{Schema: Schema, Metadata: metadata, Payload: encoded} + err = validateEnvelope(envelope) + if err != nil { + return nil, fmt.Errorf("validate envelope: %w", err) + } + + encoded, err = json.Marshal(envelope) + if err != nil { + return nil, fmt.Errorf("encode envelope: %w", err) + } + + return encoded, nil +} + +// 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 + + err := decodeObject(data, &envelope) + if err != nil { + return Envelope{}, fmt.Errorf("decode envelope: %w", err) + } + + err = validateEnvelope(envelope) + if err != nil { + return Envelope{}, fmt.Errorf("validate envelope: %w", err) + } + + return envelope, nil +} + +func validateEnvelope(envelope Envelope) error { + if envelope.Schema != Schema { + return fmt.Errorf("unsupported schema %d", envelope.Schema) + } + + if !ValidID(envelope.MessageID) { + return fmt.Errorf("message ID must be a canonical ULID") + } + + if envelope.CorrelationID != "" && !ValidID(envelope.CorrelationID) { + return fmt.Errorf("correlation ID must be a canonical ULID") + } + + if envelope.CausationID != "" && !ValidID(envelope.CausationID) { + return fmt.Errorf("causation ID must be a canonical ULID") + } + + for _, timestamp := range []*time.Time{envelope.SentAt, envelope.ExpiresAt} { + if timestamp != nil { + _, offset := timestamp.Zone() + if offset != 0 { + return fmt.Errorf("message times must use UTC") + } + } + } + + if !isObject(envelope.Payload) { + return fmt.Errorf("payload must be a JSON object") + } + + return nil +} + +func decodeObject(data []byte, target any) error { + if !isObject(data) { + return fmt.Errorf("expected one JSON object") + } + + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + + err := decoder.Decode(target) + if err != nil { + return fmt.Errorf("decode object: %w", err) + } + + var extra any + + err = decoder.Decode(&extra) + if !errors.Is(err, io.EOF) { + return fmt.Errorf("expected end of JSON object") + } + + return nil +} + +func isObject(data []byte) bool { + data = bytes.TrimSpace(data) + return len(data) > 0 && data[0] == '{' && json.Valid(data) +} diff --git a/protocol/envelope_test.go b/protocol/envelope_test.go new file mode 100644 index 0000000..dfb0600 --- /dev/null +++ b/protocol/envelope_test.go @@ -0,0 +1,61 @@ +package protocol + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +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} + data, err := EncodeEnvelope(metadata, map[string]bool{"value": true}) + require.NoError(t, err) + + envelope, err := ParseEnvelope(data) + require.NoError(t, err) + require.Equal(t, Schema, envelope.Schema) + require.Equal(t, metadata, envelope.Metadata) + require.JSONEq(t, `{"value":true}`, string(envelope.Payload)) + require.Equal(t, 5*time.Minute, MeasurementRepeatInterval) + require.Equal(t, 5*time.Minute, OnlineRepeatInterval) + require.Equal(t, 15*time.Minute, DeviceContactTimeout) +} + +func TestEnvelopeRejectsInvalidWireData(t *testing.T) { + valid := `{"schema":1,"message_id":"` + messageID + `","payload":{}}` + + testCases := []struct { + 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 + `{}`}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + _, err := ParseEnvelope([]byte(tc.data)) + require.Error(t, err) + }) + } + + _, err := EncodeEnvelope(Metadata{MessageID: "bad"}, map[string]bool{}) + require.Error(t, err) + + _, err = EncodeEnvelope(Metadata{MessageID: messageID}, make(chan int)) + require.Error(t, err) + + _, err = EncodeEnvelope(Metadata{MessageID: messageID}, []int{1}) + require.Error(t, err) +} diff --git a/protocol/identifier.go b/protocol/identifier.go new file mode 100644 index 0000000..5d3cb06 --- /dev/null +++ b/protocol/identifier.go @@ -0,0 +1,88 @@ +package protocol + +import ( + "crypto/rand" + "fmt" + "strings" + "time" +) + +const idAlphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + +// NewID returns a canonical ULID with a millisecond timestamp and 80 random +// bits. Times outside the unsigned 48-bit millisecond range are rejected. +func NewID(now time.Time) (string, error) { + milliseconds := now.UnixMilli() + if milliseconds < 0 || milliseconds >= 1<<48 { + return "", fmt.Errorf("ID time is outside the ULID range") + } + + var data [16]byte + + // Six bytes hold the timestamp and ten bytes hold random data. + for index := 5; index >= 0; index-- { + data[index] = byte(milliseconds) + milliseconds >>= 8 + } + + _, err := rand.Read(data[6:]) + if err != nil { + return "", fmt.Errorf("read ID randomness: %w", err) + } + + var text [26]byte + + for index := range text { + var value byte + + for bit := range 5 { + position := index*5 + bit - 2 + value <<= 1 + if position >= 0 { + value |= (data[position/8] >> (7 - position%8)) & 1 + } + } + + text[index] = idAlphabet[value] + } + + return string(text[:]), nil +} + +// ValidID reports whether value is a canonical uppercase ULID. +func ValidID(value string) bool { + if len(value) != 26 || value[0] > '7' { + return false + } + + for _, character := range value { + if !strings.ContainsRune(idAlphabet, character) { + return false + } + } + + return true +} + +// ValidIdentifier accepts lowercase ASCII letters and digits with single +// internal hyphen or period separators. Empty identifiers are invalid. +func ValidIdentifier(value string) bool { + separator := true + + for _, character := range value { + switch { + case character >= 'a' && character <= 'z', character >= '0' && character <= '9': + separator = false + case character == '-' || character == '.': + if separator { + return false + } + + separator = true + default: + return false + } + } + + return !separator +} diff --git a/protocol/identifier_test.go b/protocol/identifier_test.go new file mode 100644 index 0000000..7825afb --- /dev/null +++ b/protocol/identifier_test.go @@ -0,0 +1,54 @@ +package protocol + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestNewIDPreservesTimestampAndAddsRandomness(t *testing.T) { + now := time.UnixMilli(1234567890123) + first, err := NewID(now) + require.NoError(t, err) + + second, err := NewID(now) + require.NoError(t, err) + require.True(t, ValidID(first)) + require.NotEqual(t, first, second) + require.Equal(t, first[:10], second[:10]) + + var milliseconds int64 + + for _, character := range first[:10] { + for index, candidate := range idAlphabet { + if candidate == character { + milliseconds = milliseconds*32 + int64(index) + } + } + } + + require.Equal(t, now.UnixMilli(), milliseconds) + + _, err = NewID(time.UnixMilli(-1)) + require.Error(t, err) + + _, err = NewID(time.UnixMilli(1 << 48)) + require.Error(t, err) +} + +func TestIdentifiersRejectNoncanonicalText(t *testing.T) { + require.True(t, ValidID(deviceID)) + + for _, value := range []string{"", "8123456789ABCDEFGHJKMNPQRST", "0123456789abcdefghjkmnpqrst", "0123456789ABCDEFGHIKMNPQRST"} { + require.False(t, ValidID(value), value) + } + + for _, value := range []string{"relay", "a0.b1-c2"} { + require.True(t, ValidIdentifier(value), value) + } + + for _, value := range []string{"", "-relay", "relay.", "a..b", "a.-b", "a_b", "A", "+", "#"} { + require.False(t, ValidIdentifier(value), value) + } +}