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(parsePayload[DescriptorPayload](address, data, retained))
  46         case Availability:
  47                 return eraseMessage(parsePayload[AvailabilityPayload](address, data, retained))
  48         case Reported:
  49                 return eraseMessage(parsePayload[ValuePayload](address, data, retained))
  50         case Telemetry:
  51                 return eraseMessage(parsePayload[ValuePayload](address, data, retained))
  52         case Event:
  53                 return eraseMessage(parsePayload[EventPayload](address, data, retained))
  54         case Command:
  55                 return eraseMessage(parsePayload[ValuePayload](address, data, retained))
  56         case CommandBatch:
  57                 return eraseMessage(parsePayload[BatchPayload](address, data, retained))
  58         case Result:
  59                 return eraseMessage(parsePayload[ResultPayload](address, data, retained))
  60         case HubStatus:
  61                 return eraseMessage(parsePayload[HubStatusPayload](address, data, retained))
  62         default:
  63                 return Message[any]{}, fmt.Errorf("unsupported message kind")
  64         }
  65 }
  66
  67 func buildMessage[T any](address Address, metadata Metadata, payload T) (Publication, error) {
  68         topic, err := address.topic()
  69         if err != nil {
  70                 return Publication{}, fmt.Errorf("build address: %w", err)
  71         }
  72
  73         normalized := normalizePayload(payload)
  74         err = validatePayload(address, metadata, normalized)
  75         if err != nil {
  76                 return Publication{}, fmt.Errorf("validate publication: %w", err)
  77         }
  78
  79         data, err := encodeEnvelope(metadata, normalized)
  80         if err != nil {
  81                 return Publication{}, fmt.Errorf("build envelope: %w", err)
  82         }
  83
  84         retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
  85
  86         qos := byte(1)
  87         if address.Kind == Telemetry {
  88                 qos = 0
  89         }
  90
  91         publication := Publication{
  92                 Topic:  topic,
  93                 Data:   data,
  94                 QoS:    qos,
  95                 Retain: retain,
  96         }
  97
  98         switch address.Kind {
  99         case Command:
 100                 publication.ExpiresAt = metadata.ExpiresAt
 101                 publication.CorrelationData = metadata.MessageID
 102                 publication.ResponseTopic, err = (Address{
 103                         Kind:   Result,
 104                         ID:     address.ID,
 105                         Target: metadata.MessageID,
 106                 }).topic()
 107                 if err != nil {
 108                         return Publication{}, fmt.Errorf("build command response topic: %w", err)
 109                 }
 110         case CommandBatch:
 111                 batch, ok := any(payload).(BatchPayload)
 112                 if !ok {
 113                         return Publication{}, fmt.Errorf("batch requires command members")
 114                 }
 115
 116                 for _, command := range batch.Commands {
 117                         if publication.ExpiresAt == nil || command.ExpiresAt.Before(*publication.ExpiresAt) {
 118                                 publication.ExpiresAt = command.ExpiresAt
 119                         }
 120                 }
 121         case HubStatus:
 122                 publication.ExpiresAt = metadata.ExpiresAt
 123         }
 124
 125         return publication, nil
 126 }
 127
 128 func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (Message[T], error) {
 129         address, err := parseTopic(topic)
 130         if err != nil {
 131                 return Message[T]{}, fmt.Errorf("parse address: %w", err)
 132         }
 133
 134         if address.Kind != kind {
 135                 return Message[T]{}, fmt.Errorf("message topic does not match parser")
 136         }
 137
 138         message, err := parsePayload[T](address, data, retained)
 139         if err != nil {
 140                 return Message[T]{}, fmt.Errorf("parse payload: %w", err)
 141         }
 142
 143         return message, nil
 144 }
 145
 146 func eraseMessage[T any](message Message[T], err error) (Message[any], error) {
 147         if err != nil {
 148                 return Message[any]{}, fmt.Errorf("parse message: %w", err)
 149         }
 150
 151         return Message[any]{
 152                 Address:  message.Address,
 153                 Metadata: message.Metadata,
 154                 Payload:  message.Payload,
 155                 Retained: message.Retained,
 156         }, nil
 157 }
 158
 159 func parsePayload[T any](address Address, data []byte, retained bool) (Message[T], error) {
 160         kind := address.Kind
 161
 162         if retained && (kind == Telemetry || kind == Event || kind == Command || kind == CommandBatch || kind == Result) {
 163                 return Message[T]{}, fmt.Errorf("message must not be retained")
 164         }
 165
 166         envelope, err := parseEnvelope(data)
 167         if err != nil {
 168                 return Message[T]{}, fmt.Errorf("parse envelope: %w", err)
 169         }
 170
 171         var payload T
 172
 173         err = decodeObject(envelope.Payload, &payload)
 174         if err != nil {
 175                 return Message[T]{}, fmt.Errorf("decode payload: %w", err)
 176         }
 177
 178         err = validatePayload(address, envelope.Metadata, payload)
 179         if err != nil {
 180                 return Message[T]{}, fmt.Errorf("validate payload: %w", err)
 181         }
 182
 183         if address.Kind == Descriptor {
 184                 err = validateDescriptorFlags(envelope.Payload)
 185                 if err != nil {
 186                         return Message[T]{}, fmt.Errorf("validate descriptor access flags: %w", err)
 187                 }
 188         }
 189
 190         return Message[T]{
 191                 Address:  address,
 192                 Metadata: envelope.Metadata,
 193                 Payload:  payload,
 194                 Retained: retained,
 195         }, nil
 196 }
 197
 198 // normalizePayload keeps nil raw values as JSON null without changing caller
 199 // data.
 200 func normalizePayload(payload any) any {
 201         switch value := payload.(type) {
 202         case ValuePayload:
 203                 if value.Value == nil {
 204                         value.Value = json.RawMessage(`null`)
 205                 }
 206
 207                 return value
 208         case BatchPayload:
 209                 commands := make([]BatchCommand, len(value.Commands))
 210                 copy(commands, value.Commands)
 211
 212                 for index := range commands {
 213                         if commands[index].Payload.Value == nil {
 214                                 commands[index].Payload.Value = json.RawMessage(`null`)
 215                         }
 216                 }
 217
 218                 value.Commands = commands
 219
 220                 return value
 221         default:
 222                 return payload
 223         }
 224 }
 225
 226 func validatePayload(address Address, metadata Metadata, payload any) error {
 227         var err error
 228
 229         switch value := payload.(type) {
 230         case DescriptorPayload:
 231                 err = validateDescriptor(value)
 232         case AvailabilityPayload:
 233                 err = validateAvailability(value)
 234         case ValuePayload:
 235                 err = validateValue(value, address.Kind, metadata)
 236         case EventPayload:
 237                 err = validateEvent(value)
 238         case ResultPayload:
 239                 err = validateResult(value, address, metadata)
 240         case BatchPayload:
 241                 err = validateBatch(value, metadata)
 242         case HubStatusPayload:
 243                 err = validateHub(value, address, metadata)
 244         default:
 245                 return fmt.Errorf("unsupported payload type")
 246         }
 247
 248         if err != nil {
 249                 return fmt.Errorf("validate %s payload: %w", address.Kind, err)
 250         }
 251
 252         return nil
 253 }
 254
 255 func validateDescriptorFlags(raw json.RawMessage) error {
 256         var fields struct {
 257                 Capabilities []map[string]json.RawMessage `json:"capabilities"`
 258         }
 259
 260         err := json.Unmarshal(raw, &fields)
 261         if err != nil {
 262                 return fmt.Errorf("read descriptor access flags: %w", err)
 263         }
 264
 265         for _, capability := range fields.Capabilities {
 266                 for _, name := range []string{"reports_state", "accepts_commands"} {
 267                         field := capability[name]
 268                         if string(field) != "true" && string(field) != "false" {
 269                                 return fmt.Errorf("capability access flags must be present booleans")
 270                         }
 271                 }
 272         }
 273
 274         return nil
 275 }
 276
 277 func validateDescriptor(payload DescriptorPayload) error {
 278         if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(string([]rune(payload.Attribution))) > 2000 || len(payload.Capabilities) == 0 {
 279                 return fmt.Errorf("descriptor metadata or capabilities are invalid")
 280         }
 281
 282         seen := make(map[string]bool, len(payload.Capabilities))
 283
 284         for _, capability := range payload.Capabilities {
 285                 if !ValidIdentifier(capability.ID) || !ValidIdentifier(capability.Type) || seen[capability.ID] {
 286                         return fmt.Errorf("capability IDs and types must be valid and unique")
 287                 }
 288
 289                 if !capability.ReportsState && !capability.AcceptsCommands {
 290                         return fmt.Errorf("capability must have at least one access mode")
 291                 }
 292
 293                 seen[capability.ID] = true
 294         }
 295
 296         return nil
 297 }
 298
 299 func validateAvailability(payload AvailabilityPayload) error {
 300         if payload.Status != "online" && payload.Status != "offline" {
 301                 return fmt.Errorf("availability must be online or offline")
 302         }
 303
 304         return nil
 305 }
 306
 307 func validateValue(payload ValuePayload, kind Kind, metadata Metadata) error {
 308         if !json.Valid(payload.Value) {
 309                 return fmt.Errorf("value is required and must be valid JSON")
 310         }
 311
 312         if kind == Command && metadata.ExpiresAt == nil {
 313                 return fmt.Errorf("command must have an expiry")
 314         }
 315
 316         return nil
 317 }
 318
 319 func validateEvent(payload EventPayload) error {
 320         if payload == nil {
 321                 return fmt.Errorf("event must be a JSON object")
 322         }
 323
 324         for _, value := range payload {
 325                 if value != nil && !json.Valid(value) {
 326                         return fmt.Errorf("event values must be valid JSON")
 327                 }
 328         }
 329
 330         return nil
 331 }
 332
 333 func validateResult(payload ResultPayload, address Address, metadata Metadata) error {
 334         if metadata.CorrelationID != address.Target {
 335                 return fmt.Errorf("result correlation must match its command ID")
 336         }
 337
 338         switch payload.Status {
 339         case "accepted", "completed":
 340                 if payload.Reason != "" {
 341                         return fmt.Errorf("successful result must not have a reason")
 342                 }
 343         case "rejected", "failed":
 344                 if payload.Reason == "" {
 345                         return fmt.Errorf("unsuccessful result must have a reason")
 346                 }
 347         default:
 348                 return fmt.Errorf("result status is invalid")
 349         }
 350
 351         return nil
 352 }
 353
 354 func validateBatch(payload BatchPayload, metadata Metadata) error {
 355         if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
 356                 return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
 357         }
 358
 359         if metadata.MessageID != payload.Commands[0].MessageID {
 360                 return fmt.Errorf("batch ID must equal its first command ID")
 361         }
 362
 363         ids := make(map[string]bool, len(payload.Commands))
 364         targets := make(map[string]bool, len(payload.Commands))
 365
 366         for _, command := range payload.Commands {
 367                 if !ValidIdentifier(command.Capability) || ids[command.MessageID] || targets[command.Capability] {
 368                         return fmt.Errorf("batch targets and IDs must be valid and distinct")
 369                 }
 370
 371                 err := validateMetadata(command.Metadata)
 372                 if err != nil {
 373                         return fmt.Errorf("validate batch command metadata: %w", err)
 374                 }
 375
 376                 err = validateValue(command.Payload, Command, command.Metadata)
 377                 if err != nil {
 378                         return fmt.Errorf("validate batch command value: %w", err)
 379                 }
 380
 381                 ids[command.MessageID] = true
 382                 targets[command.Capability] = true
 383         }
 384
 385         return nil
 386 }
 387
 388 func validateHub(payload HubStatusPayload, address Address, metadata Metadata) error {
 389         if payload.ID != address.ID || metadata.ExpiresAt == nil {
 390                 return fmt.Errorf("hub status requires a matching ID and an expiry")
 391         }
 392
 393         if payload.Status != "ready" && payload.Status != "unavailable" {
 394                 return fmt.Errorf("hub status must be ready or unavailable")
 395         }
 396
 397         if payload.Address == "" || payload.ConfiguredDevices < 0 || payload.PairedDevices != nil && *payload.PairedDevices < 0 {
 398                 return fmt.Errorf("hub address or device counts are invalid")
 399         }
 400
 401         timestamps := []*time.Time{payload.UpdatedAt}
 402         if payload.Diagnostics != nil {
 403                 timestamps = append(timestamps, &payload.Diagnostics.ObservedAt)
 404         }
 405
 406         if payload.Pairing != nil {
 407                 timestamps = append(timestamps, &payload.Pairing.RequestedAt, &payload.Pairing.Deadline)
 408         }
 409
 410         for _, timestamp := range timestamps {
 411                 if timestamp != nil {
 412                         _, offset := timestamp.Zone()
 413                         if offset != 0 || timestamp.IsZero() {
 414                                 return fmt.Errorf("hub timestamps must be nonzero UTC times")
 415                         }
 416                 }
 417         }
 418
 419         return nil
 420 }