git.giacomolli.uk - onehouse/blob - driver/publication_value.go

   1 package driver
   2
   3 import (
   4         "bytes"
   5         "encoding/json"
   6         "fmt"
   7         "time"
   8
   9         "onehouse/protocol"
  10 )
  11
  12 type publicationValue struct {
  13         publication protocol.Publication
  14         address     protocol.Address
  15         metadata    protocol.Metadata
  16         payload     any
  17         comparison  []byte
  18         measurement bool
  19         online      bool
  20 }
  21
  22 func newPublicationValue[T any](address protocol.Address, metadata protocol.Metadata, payload T) (publicationValue, error) {
  23         metadata, err := newMetadata(metadata)
  24         if err != nil {
  25                 return publicationValue{}, fmt.Errorf("create publication metadata: %w", err)
  26         }
  27
  28         value := publicationValue{
  29                 address:  address,
  30                 metadata: metadata,
  31                 payload:  payload,
  32         }
  33         value.publication, err = value.build()
  34         if err != nil {
  35                 return publicationValue{}, fmt.Errorf("build typed publication: %w", err)
  36         }
  37
  38         data, err := json.Marshal(payload)
  39         if err != nil {
  40                 return publicationValue{}, fmt.Errorf("encode comparison payload: %w", err)
  41         }
  42
  43         var normalized any
  44
  45         decoder := json.NewDecoder(bytes.NewReader(data))
  46         decoder.UseNumber()
  47
  48         err = decoder.Decode(&normalized)
  49         if err != nil {
  50                 return publicationValue{}, fmt.Errorf("decode comparison payload: %w", err)
  51         }
  52
  53         comparison, err := json.Marshal(normalized)
  54         if err != nil {
  55                 return publicationValue{}, fmt.Errorf("normalize comparison payload: %w", err)
  56         }
  57
  58         var copied T
  59
  60         err = json.Unmarshal(comparison, &copied)
  61         if err != nil {
  62                 return publicationValue{}, fmt.Errorf("copy publication payload: %w", err)
  63         }
  64
  65         value.payload = copied
  66         value.comparison = comparison
  67
  68         switch typed := any(copied).(type) {
  69         case protocol.ValuePayload:
  70                 value.measurement = address.Kind == protocol.Telemetry || string(typed.Value) != "true" && string(typed.Value) != "false"
  71         case protocol.AvailabilityPayload:
  72                 value.online = typed.Status == "online"
  73         }
  74
  75         return value, nil
  76 }
  77
  78 func newMetadata(metadata protocol.Metadata) (protocol.Metadata, error) {
  79         now := time.Now().UTC()
  80         if metadata.MessageID == "" {
  81                 id, err := protocol.NewID(now)
  82                 if err != nil {
  83                         return protocol.Metadata{}, fmt.Errorf("create message ID: %w", err)
  84                 }
  85
  86                 metadata.MessageID = id
  87         }
  88
  89         if metadata.SentAt == nil {
  90                 metadata.SentAt = &now
  91         } else {
  92                 timestamp := *metadata.SentAt
  93                 metadata.SentAt = &timestamp
  94         }
  95
  96         if metadata.ExpiresAt != nil {
  97                 timestamp := *metadata.ExpiresAt
  98                 metadata.ExpiresAt = &timestamp
  99         }
 100
 101         return metadata, nil
 102 }
 103
 104 func (v publicationValue) build() (protocol.Publication, error) {
 105         var (
 106                 publication protocol.Publication
 107                 err         error
 108         )
 109
 110         switch payload := v.payload.(type) {
 111         case protocol.DescriptorPayload:
 112                 publication, err = protocol.BuildDescriptor(v.address.ID, v.metadata, payload)
 113         case protocol.AvailabilityPayload:
 114                 publication, err = protocol.BuildAvailability(v.address.ID, v.metadata, payload)
 115         case protocol.ValuePayload:
 116                 if v.address.Kind == protocol.Telemetry {
 117                         publication, err = protocol.BuildTelemetry(v.address.ID, v.address.Target, v.metadata, payload)
 118                 } else {
 119                         publication, err = protocol.BuildReported(v.address.ID, v.address.Target, v.metadata, payload)
 120                 }
 121         case protocol.EventPayload:
 122                 publication, err = protocol.BuildEvent(v.address.ID, v.address.Target, v.metadata, payload)
 123         case protocol.ResultPayload:
 124                 publication, err = protocol.BuildResult(v.address.ID, v.address.Target, v.metadata, payload)
 125         default:
 126                 return protocol.Publication{}, fmt.Errorf("unsupported publication payload")
 127         }
 128
 129         if err != nil {
 130                 return protocol.Publication{}, fmt.Errorf("build %s: %w", v.address.Kind, err)
 131         }
 132
 133         return publication, nil
 134 }
 135
 136 func (v publicationValue) repeated() (publicationValue, error) {
 137         metadata, err := newMetadata(protocol.Metadata{SentAt: v.metadata.SentAt})
 138         if err != nil {
 139                 return publicationValue{}, fmt.Errorf("create repeat metadata: %w", err)
 140         }
 141
 142         v.metadata = metadata
 143         if v.address.Kind == protocol.Telemetry {
 144                 v.address.Kind = protocol.Reported
 145         }
 146
 147         v.publication, err = v.build()
 148         if err != nil {
 149                 return publicationValue{}, fmt.Errorf("build repeated publication: %w", err)
 150         }
 151
 152         return v, nil
 153 }