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