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         if !ValidID(envelope.MessageID) {
  81                 return fmt.Errorf("message ID must be a canonical ULID")
  82         }
  83
  84         if envelope.CorrelationID != "" && !ValidID(envelope.CorrelationID) {
  85                 return fmt.Errorf("correlation ID must be a canonical ULID")
  86         }
  87
  88         if envelope.CausationID != "" && !ValidID(envelope.CausationID) {
  89                 return fmt.Errorf("causation ID must be a canonical ULID")
  90         }
  91
  92         for _, timestamp := range []*time.Time{envelope.SentAt, envelope.ExpiresAt} {
  93                 if timestamp != nil {
  94                         _, offset := timestamp.Zone()
  95                         if offset != 0 {
  96                                 return fmt.Errorf("message times must use UTC")
  97                         }
  98                 }
  99         }
 100
 101         if !isObject(envelope.Payload) {
 102                 return fmt.Errorf("payload must be a JSON object")
 103         }
 104
 105         return nil
 106 }
 107
 108 func decodeObject(data []byte, target any) error {
 109         if !isObject(data) {
 110                 return fmt.Errorf("expected one JSON object")
 111         }
 112
 113         decoder := json.NewDecoder(bytes.NewReader(data))
 114         decoder.DisallowUnknownFields()
 115
 116         err := decoder.Decode(target)
 117         if err != nil {
 118                 return fmt.Errorf("decode object: %w", err)
 119         }
 120
 121         var extra any
 122
 123         err = decoder.Decode(&extra)
 124         if !errors.Is(err, io.EOF) {
 125                 return fmt.Errorf("expected end of JSON object")
 126         }
 127
 128         return nil
 129 }
 130
 131 func isObject(data []byte) bool {
 132         data = bytes.TrimSpace(data)
 133         return len(data) > 0 && data[0] == '{' && json.Valid(data)
 134 }