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

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