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