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

   1 package protocol
   2
   3 import (
   4         "bytes"
   5         "encoding/json"
   6         "errors"
   7         "fmt"
   8         "io"
   9         "time"
  10 )
  11
  12 const (
  13         // Schema is the supported envelope and payload schema version.
  14         Schema = 1
  15
  16         // MeasurementRepeatInterval bounds equal retained measurement repeats.
  17         MeasurementRepeatInterval = 5 * time.Minute
  18         // OnlineRepeatInterval bounds equal online availability repeats.
  19         OnlineRepeatInterval = 5 * time.Minute
  20         // DeviceContactTimeout bounds the age of device contact evidence.
  21         DeviceContactTimeout = 15 * time.Minute
  22 )
  23
  24 // Metadata holds message identity and optional UTC times. Retries keep the same
  25 // ID. A receiver checks ExpiresAt against its clock before processing.
  26 type Metadata struct {
  27         MessageID     string     `json:"message_id"`
  28         SentAt        *time.Time `json:"sent_at,omitempty"`
  29         ExpiresAt     *time.Time `json:"expires_at,omitempty"`
  30         CorrelationID string     `json:"correlation_id,omitempty"`
  31         CausationID   string     `json:"causation_id,omitempty"`
  32 }
  33
  34 // Envelope carries one JSON object and its schema and metadata.
  35 type Envelope struct {
  36         Schema int `json:"schema"`
  37         Metadata
  38         Payload json.RawMessage `json:"payload"`
  39 }
  40
  41 // EncodeEnvelope validates metadata and encodes a schema-1 object payload.
  42 func EncodeEnvelope(metadata Metadata, payload any) ([]byte, error) {
  43         encoded, err := json.Marshal(payload)
  44         if err != nil {
  45                 return nil, fmt.Errorf("encode payload: %w", err)
  46         }
  47
  48         envelope := Envelope{Schema: Schema, Metadata: metadata, Payload: encoded}
  49         err = validateEnvelope(envelope)
  50         if err != nil {
  51                 return nil, fmt.Errorf("validate envelope: %w", err)
  52         }
  53
  54         encoded, err = json.Marshal(envelope)
  55         if err != nil {
  56                 return nil, fmt.Errorf("encode envelope: %w", err)
  57         }
  58
  59         return encoded, nil
  60 }
  61
  62 // ParseEnvelope rejects unknown fields, trailing JSON, unsupported schemas,
  63 // invalid IDs, non-UTC times, and missing or non-object payloads.
  64 func ParseEnvelope(data []byte) (Envelope, error) {
  65         var envelope Envelope
  66
  67         err := decodeObject(data, &envelope)
  68         if err != nil {
  69                 return Envelope{}, fmt.Errorf("decode envelope: %w", err)
  70         }
  71
  72         err = validateEnvelope(envelope)
  73         if err != nil {
  74                 return Envelope{}, fmt.Errorf("validate envelope: %w", err)
  75         }
  76
  77         return envelope, nil
  78 }
  79
  80 func validateEnvelope(envelope Envelope) error {
  81         if envelope.Schema != Schema {
  82                 return fmt.Errorf("unsupported schema %d", envelope.Schema)
  83         }
  84
  85         if !ValidID(envelope.MessageID) {
  86                 return fmt.Errorf("message ID must be a canonical ULID")
  87         }
  88
  89         if envelope.CorrelationID != "" && !ValidID(envelope.CorrelationID) {
  90                 return fmt.Errorf("correlation ID must be a canonical ULID")
  91         }
  92
  93         if envelope.CausationID != "" && !ValidID(envelope.CausationID) {
  94                 return fmt.Errorf("causation ID must be a canonical ULID")
  95         }
  96
  97         for _, timestamp := range []*time.Time{envelope.SentAt, envelope.ExpiresAt} {
  98                 if timestamp != nil {
  99                         _, offset := timestamp.Zone()
 100                         if offset != 0 {
 101                                 return fmt.Errorf("message times must use UTC")
 102                         }
 103                 }
 104         }
 105
 106         if !isObject(envelope.Payload) {
 107                 return fmt.Errorf("payload must be a JSON object")
 108         }
 109
 110         return nil
 111 }
 112
 113 func decodeObject(data []byte, target any) error {
 114         if !isObject(data) {
 115                 return fmt.Errorf("expected one JSON object")
 116         }
 117
 118         decoder := json.NewDecoder(bytes.NewReader(data))
 119         decoder.DisallowUnknownFields()
 120
 121         err := decoder.Decode(target)
 122         if err != nil {
 123                 return fmt.Errorf("decode object: %w", err)
 124         }
 125
 126         var extra any
 127
 128         err = decoder.Decode(&extra)
 129         if !errors.Is(err, io.EOF) {
 130                 return fmt.Errorf("expected end of JSON object")
 131         }
 132
 133         return nil
 134 }
 135
 136 func isObject(data []byte) bool {
 137         data = bytes.TrimSpace(data)
 138         return len(data) > 0 && data[0] == '{' && json.Valid(data)
 139 }