git.giacomolli.uk - onehouse/blob - protocol/message.go

   1 package protocol
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "time"
   7 )
   8
   9 // Message contains validated routing, metadata, and a typed payload. Retained
  10 // is the delivery flag. Live delivery of retained-topic data can be false.
  11 type Message[T any] struct {
  12         Address  Address
  13         Metadata Metadata
  14         Payload  T
  15         Retained bool
  16 }
  17
  18 // Publication is a broker-independent outgoing message. The transport sets
  19 // UTF-8 payload format and application/json content type. Commands and hub
  20 // status also need MQTT expiry computed from their envelope deadlines.
  21 type Publication struct {
  22         Topic  string
  23         Data   []byte
  24         QoS    byte
  25         Retain bool
  26         // ExpiresAt is the command, earliest batch member, or hub deadline.
  27         ExpiresAt *time.Time
  28         // ResponseTopic and CorrelationData apply to individual commands only.
  29         ResponseTopic   string
  30         CorrelationData string
  31 }
  32
  33 // Parse routes an incoming publication to a typed payload without caller topic
  34 // operations. Payload is DescriptorPayload, AvailabilityPayload, ValuePayload,
  35 // EventPayload, BatchPayload, ResultPayload, or HubStatusPayload. Address.Kind
  36 // distinguishes command, reported, and telemetry value payloads.
  37 func Parse(topic string, data []byte, retained bool) (Message[any], error) {
  38         address, err := parseTopic(topic)
  39         if err != nil {
  40                 return Message[any]{}, fmt.Errorf("parse address: %w", err)
  41         }
  42
  43         switch address.Kind {
  44         case Descriptor:
  45                 return eraseMessage(ParseDescriptor(topic, data, retained))
  46         case Availability:
  47                 return eraseMessage(ParseAvailability(topic, data, retained))
  48         case Reported:
  49                 return eraseMessage(ParseReported(topic, data, retained))
  50         case Telemetry:
  51                 return eraseMessage(ParseTelemetry(topic, data, retained))
  52         case Event:
  53                 return eraseMessage(ParseEvent(topic, data, retained))
  54         case Command:
  55                 return eraseMessage(ParseCommand(topic, data, retained))
  56         case CommandBatch:
  57                 return eraseMessage(ParseCommandBatch(topic, data, retained))
  58         case Result:
  59                 return eraseMessage(ParseResult(topic, data, retained))
  60         case HubStatus:
  61                 return eraseMessage(ParseHubStatus(topic, data, retained))
  62         default:
  63                 return Message[any]{}, fmt.Errorf("unsupported message kind")
  64         }
  65 }
  66
  67 func eraseMessage[T any](message Message[T], err error) (Message[any], error) {
  68         if err != nil {
  69                 return Message[any]{}, fmt.Errorf("parse message: %w", err)
  70         }
  71
  72         return Message[any]{
  73                 Address:  message.Address,
  74                 Metadata: message.Metadata,
  75                 Payload:  message.Payload,
  76                 Retained: message.Retained,
  77         }, nil
  78 }
  79
  80 func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) {
  81         topic, err := address.topic()
  82         if err != nil {
  83                 return Publication{}, fmt.Errorf("build address: %w", err)
  84         }
  85
  86         data, err := encodeEnvelope(metadata, payload)
  87         if err != nil {
  88                 return Publication{}, fmt.Errorf("build envelope: %w", err)
  89         }
  90
  91         retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
  92         _, err = parseMessage[T](address.Kind, topic, data, retain)
  93         if err != nil {
  94                 return Publication{}, fmt.Errorf("validate publication: %w", err)
  95         }
  96
  97         qos := byte(1)
  98         if address.Kind == Telemetry {
  99                 qos = 0
 100         }
 101
 102         publication := Publication{
 103                 Topic:  topic,
 104                 Data:   data,
 105                 QoS:    qos,
 106                 Retain: retain,
 107         }
 108
 109         switch address.Kind {
 110         case Command:
 111                 publication.ExpiresAt = metadata.ExpiresAt
 112                 publication.CorrelationData = metadata.MessageID
 113                 publication.ResponseTopic, err = (Address{
 114                         Kind:   Result,
 115                         ID:     address.ID,
 116                         Target: metadata.MessageID,
 117                 }).topic()
 118                 if err != nil {
 119                         return Publication{}, fmt.Errorf("build command response topic: %w", err)
 120                 }
 121         case CommandBatch:
 122                 batch, ok := any(payload).(BatchPayload)
 123                 if !ok {
 124                         return Publication{}, fmt.Errorf("batch requires command members")
 125                 }
 126
 127                 for _, command := range batch.Commands {
 128                         if publication.ExpiresAt == nil || command.ExpiresAt.Before(*publication.ExpiresAt) {
 129                                 publication.ExpiresAt = command.ExpiresAt
 130                         }
 131                 }
 132         case HubStatus:
 133                 publication.ExpiresAt = metadata.ExpiresAt
 134         }
 135
 136         return publication, nil
 137 }
 138
 139 func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (Message[T], error) {
 140         address, err := parseTopic(topic)
 141         if err != nil {
 142                 return Message[T]{}, fmt.Errorf("parse address: %w", err)
 143         }
 144
 145         if address.Kind != kind {
 146                 return Message[T]{}, fmt.Errorf("message topic does not match parser")
 147         }
 148
 149         if retained && (kind == Telemetry || kind == Event || kind == Command || kind == CommandBatch || kind == Result) {
 150                 return Message[T]{}, fmt.Errorf("message must not be retained")
 151         }
 152
 153         envelope, err := parseEnvelope(data)
 154         if err != nil {
 155                 return Message[T]{}, fmt.Errorf("parse envelope: %w", err)
 156         }
 157
 158         var payload T
 159
 160         err = decodeObject(envelope.Payload, &payload)
 161         if err != nil {
 162                 return Message[T]{}, fmt.Errorf("decode payload: %w", err)
 163         }
 164
 165         err = validatePayload(address, envelope, payload)
 166         if err != nil {
 167                 return Message[T]{}, fmt.Errorf("validate payload: %w", err)
 168         }
 169
 170         return Message[T]{
 171                 Address:  address,
 172                 Metadata: envelope.Metadata,
 173                 Payload:  payload,
 174                 Retained: retained,
 175         }, nil
 176 }
 177
 178 func validatePayload(address Address, envelope wireEnvelope, payload any) error {
 179         switch value := payload.(type) {
 180         case DescriptorPayload:
 181                 err := validateDescriptor(value, envelope.Payload)
 182                 if err != nil {
 183                         return fmt.Errorf("validate descriptor: %w", err)
 184                 }
 185         case AvailabilityPayload:
 186                 if value.Status != "online" && value.Status != "offline" {
 187                         return fmt.Errorf("availability must be online or offline")
 188                 }
 189         case ValuePayload:
 190                 if !json.Valid(value.Value) {
 191                         return fmt.Errorf("value is required and must be valid JSON")
 192                 }
 193
 194                 if address.Kind == Command && envelope.ExpiresAt == nil {
 195                         return fmt.Errorf("command must have an expiry")
 196                 }
 197         case EventPayload:
 198                 // An event object has application-defined fields.
 199         case ResultPayload:
 200                 if envelope.CorrelationID != address.Target {
 201                         return fmt.Errorf("result correlation must match its command ID")
 202                 }
 203
 204                 switch value.Status {
 205                 case "accepted", "completed":
 206                         if value.Reason != "" {
 207                                 return fmt.Errorf("successful result must not have a reason")
 208                         }
 209                 case "rejected", "failed":
 210                         if value.Reason == "" {
 211                                 return fmt.Errorf("unsuccessful result must have a reason")
 212                         }
 213                 default:
 214                         return fmt.Errorf("result status is invalid")
 215                 }
 216         case BatchPayload:
 217                 err := validateBatch(value, envelope)
 218                 if err != nil {
 219                         return fmt.Errorf("validate command batch: %w", err)
 220                 }
 221         case HubStatusPayload:
 222                 err := validateHub(value, address, envelope)
 223                 if err != nil {
 224                         return fmt.Errorf("validate hub status: %w", err)
 225                 }
 226         default:
 227                 return fmt.Errorf("unsupported payload type")
 228         }
 229
 230         return nil
 231 }
 232
 233 func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
 234         if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(payload.Attribution) > 2000 || len(payload.Capabilities) == 0 {
 235                 return fmt.Errorf("descriptor metadata or capabilities are invalid")
 236         }
 237
 238         var fields struct {
 239                 Capabilities []map[string]json.RawMessage `json:"capabilities"`
 240         }
 241
 242         err := json.Unmarshal(raw, &fields)
 243         if err != nil {
 244                 return fmt.Errorf("read descriptor access flags: %w", err)
 245         }
 246
 247         seen := make(map[string]bool, len(payload.Capabilities))
 248
 249         for index, capability := range payload.Capabilities {
 250                 if !ValidIdentifier(capability.ID) || !ValidIdentifier(capability.Type) || seen[capability.ID] {
 251                         return fmt.Errorf("capability IDs and types must be valid and unique")
 252                 }
 253
 254                 if !capability.ReportsState && !capability.AcceptsCommands {
 255                         return fmt.Errorf("capability must have at least one access mode")
 256                 }
 257
 258                 for _, name := range []string{"reports_state", "accepts_commands"} {
 259                         field := fields.Capabilities[index][name]
 260                         if string(field) != "true" && string(field) != "false" {
 261                                 return fmt.Errorf("capability access flags must be present booleans")
 262                         }
 263                 }
 264
 265                 seen[capability.ID] = true
 266         }
 267
 268         return nil
 269 }
 270
 271 func validateBatch(payload BatchPayload, envelope wireEnvelope) error {
 272         if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
 273                 return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
 274         }
 275
 276         if envelope.MessageID != payload.Commands[0].MessageID {
 277                 return fmt.Errorf("batch ID must equal its first command ID")
 278         }
 279
 280         ids := make(map[string]bool, len(payload.Commands))
 281         targets := make(map[string]bool, len(payload.Commands))
 282
 283         for _, command := range payload.Commands {
 284                 if !ValidIdentifier(command.Capability) || ids[command.MessageID] || targets[command.Capability] {
 285                         return fmt.Errorf("batch targets and IDs must be valid and distinct")
 286                 }
 287
 288                 _, err := encodeEnvelope(command.Metadata, command.Payload)
 289                 if err != nil {
 290                         return fmt.Errorf("validate batch command envelope: %w", err)
 291                 }
 292
 293                 if command.ExpiresAt == nil || !json.Valid(command.Payload.Value) {
 294                         return fmt.Errorf("batch commands require expiry and a JSON value")
 295                 }
 296
 297                 ids[command.MessageID] = true
 298                 targets[command.Capability] = true
 299         }
 300
 301         return nil
 302 }
 303
 304 func validateHub(payload HubStatusPayload, address Address, envelope wireEnvelope) error {
 305         if payload.ID != address.ID || envelope.ExpiresAt == nil {
 306                 return fmt.Errorf("hub status requires a matching ID and an expiry")
 307         }
 308
 309         if payload.Status != "ready" && payload.Status != "unavailable" {
 310                 return fmt.Errorf("hub status must be ready or unavailable")
 311         }
 312
 313         if payload.Address == "" || payload.ConfiguredDevices < 0 || payload.PairedDevices != nil && *payload.PairedDevices < 0 {
 314                 return fmt.Errorf("hub address or device counts are invalid")
 315         }
 316
 317         timestamps := []*time.Time{payload.UpdatedAt}
 318         if payload.Diagnostics != nil {
 319                 timestamps = append(timestamps, &payload.Diagnostics.ObservedAt)
 320         }
 321
 322         if payload.Pairing != nil {
 323                 timestamps = append(timestamps, &payload.Pairing.RequestedAt, &payload.Pairing.Deadline)
 324         }
 325
 326         for _, timestamp := range timestamps {
 327                 if timestamp != nil {
 328                         _, offset := timestamp.Zone()
 329                         if offset != 0 || timestamp.IsZero() {
 330                                 return fmt.Errorf("hub timestamps must be nonzero UTC times")
 331                         }
 332                 }
 333         }
 334
 335         return nil
 336 }