--- /dev/null
+###############
+ 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.
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
+)
--- /dev/null
+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=
inherit src version;
subPackages = [ "cmd/onehouse" ];
- vendorHash = null;
+ vendorHash = "sha256-yqtj+zCo7u2UwaQ12bHCHPNucgQNkvkN7nfkLynB67Y=";
doInstallCheck = true;
installCheckPhase = ''
--- /dev/null
+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}
+}
--- /dev/null
+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)
+}
--- /dev/null
+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)
+}
--- /dev/null
+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)
+}
--- /dev/null
+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
+}
--- /dev/null
+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)
+ }
+}
projects / onehouse / commitdiff