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

   1 package driver
   2
   3 import (
   4         "bytes"
   5         "context"
   6         "fmt"
   7         "sync"
   8         "time"
   9
  10         "onehouse/protocol"
  11 )
  12
  13 const publishTimeout = 5 * time.Second
  14
  15 // Publisher sends typed device messages and owns retained-value suppression.
  16 // Its methods are safe for concurrent use. Run owns repeats and Refresh
  17 // requests reconnect snapshots. The caller supplies a valid transport and runs
  18 // this publisher once alongside the integration drivers.
  19 type Publisher struct {
  20         transport Transport
  21         mu        sync.Mutex
  22         retained  map[protocol.Address]*retainedValue
  23         refresh   chan struct{}
  24 }
  25
  26 // NewPublisher creates a publisher with no cached state. The caller retains
  27 // ownership of transport and must keep it available until Run stops.
  28 func NewPublisher(transport Transport) *Publisher {
  29         return &Publisher{
  30                 transport: transport,
  31                 retained:  make(map[protocol.Address]*retainedValue),
  32                 refresh:   make(chan struct{}, 1),
  33         }
  34 }
  35
  36 // Run repeats measurements and online evidence at the contract interval. A send
  37 // failure returns to the runner. Observation times stay unchanged and repeated
  38 // messages have fresh IDs without command correlation or causation.
  39 func (p *Publisher) Run(ctx context.Context) error {
  40         ticker := time.NewTicker(250 * time.Millisecond)
  41         defer ticker.Stop()
  42
  43         for {
  44                 force := false
  45
  46                 select {
  47                 case <-ctx.Done():
  48                         return ctx.Err()
  49                 case <-p.refresh:
  50                         force = true
  51                 case <-ticker.C:
  52                 }
  53
  54                 err := p.repeat(ctx, force)
  55                 if err != nil {
  56                         return fmt.Errorf("repeat snapshots: %w", err)
  57                 }
  58         }
  59 }
  60
  61 // Refresh coalesces a nonblocking request to resend all cached snapshots.
  62 func (p *Publisher) Refresh() {
  63         select {
  64         case p.refresh <- struct{}{}:
  65         default:
  66         }
  67 }
  68
  69 // Descriptor sends retained metadata unless its value equals the last success.
  70 // Missing message ID and observation time are filled by the publisher. A
  71 // supplied correlation or causation always bypasses value suppression.
  72 func (p *Publisher) Descriptor(ctx context.Context, deviceID string, metadata protocol.Metadata, payload protocol.DescriptorPayload) error {
  73         return publishTyped(ctx, p, protocol.Address{
  74                 Kind: protocol.Descriptor,
  75                 ID:   deviceID,
  76         }, metadata, payload)
  77 }
  78
  79 // Availability sends reachability and repeats online evidence during Run.
  80 // Metadata defaults and command correlation follow Descriptor.
  81 func (p *Publisher) Availability(ctx context.Context, deviceID string, metadata protocol.Metadata, payload protocol.AvailabilityPayload) error {
  82         return publishTyped(ctx, p, protocol.Address{
  83                 Kind: protocol.Availability,
  84                 ID:   deviceID,
  85         }, metadata, payload)
  86 }
  87
  88 // Reported sends a complete retained value. Non-Boolean values repeat during
  89 // Run. A Boolean point also repeats after Telemetry marks it as a measurement.
  90 // Metadata defaults and command correlation follow Descriptor.
  91 func (p *Publisher) Reported(ctx context.Context, deviceID string, capability string, metadata protocol.Metadata, payload protocol.ValuePayload) error {
  92         return publishTyped(ctx, p, protocol.Address{
  93                 Kind:   protocol.Reported,
  94                 ID:     deviceID,
  95                 Target: capability,
  96         }, metadata, payload)
  97 }
  98
  99 // Telemetry sends changed measurements at QoS 0 without retention. The latest
 100 // observation supplies the retained reported snapshot on repeat or reconnect,
 101 // not on every telemetry change. Equal samples are suppressed until repeat.
 102 // Metadata defaults and command correlation follow Descriptor.
 103 func (p *Publisher) Telemetry(ctx context.Context, deviceID string, capability string, metadata protocol.Metadata, payload protocol.ValuePayload) error {
 104         return publishTyped(ctx, p, protocol.Address{
 105                 Kind:   protocol.Telemetry,
 106                 ID:     deviceID,
 107                 Target: capability,
 108         }, metadata, payload)
 109 }
 110
 111 // Event sends each occurrence without suppression. Metadata defaults follow
 112 // Descriptor. The caller owns event-specific observation and safety rules.
 113 func (p *Publisher) Event(ctx context.Context, deviceID string, event string, metadata protocol.Metadata, payload protocol.EventPayload) error {
 114         return publishTyped(ctx, p, protocol.Address{
 115                 Kind:   protocol.Event,
 116                 ID:     deviceID,
 117                 Target: event,
 118         }, metadata, payload)
 119 }
 120
 121 // Result sends an outcome without suppression. Correlation must equal the
 122 // command ID. Reuse ledger metadata to replay the same result ID and time.
 123 // Missing message ID and observation time follow Descriptor defaults.
 124 func (p *Publisher) Result(ctx context.Context, deviceID string, commandID string, metadata protocol.Metadata, payload protocol.ResultPayload) error {
 125         return publishTyped(ctx, p, protocol.Address{
 126                 Kind:   protocol.Result,
 127                 ID:     deviceID,
 128                 Target: commandID,
 129         }, metadata, payload)
 130 }
 131
 132 func (p *Publisher) publish(ctx context.Context, value publicationValue) error {
 133         err := ctx.Err()
 134         if err != nil {
 135                 return err
 136         }
 137
 138         if value.metadata.ExpiresAt != nil && !time.Now().Before(*value.metadata.ExpiresAt) {
 139                 return fmt.Errorf("publication expired")
 140         }
 141
 142         publication := value.publication
 143
 144         if !publication.Retain && value.address.Kind != protocol.Telemetry || value.metadata.ExpiresAt != nil {
 145                 err = p.send(ctx, publication)
 146                 if err != nil {
 147                         return fmt.Errorf("send direct publication: %w", err)
 148                 }
 149
 150                 return nil
 151         }
 152
 153         key := value.address
 154         if key.Kind == protocol.Telemetry {
 155                 key.Kind = protocol.Reported
 156         }
 157
 158         p.mu.Lock()
 159
 160         entry := p.retained[key]
 161         if entry == nil {
 162                 entry = &retainedValue{retainedAt: time.Now()}
 163                 p.retained[key] = entry
 164         }
 165
 166         p.mu.Unlock()
 167         entry.mu.Lock()
 168         defer entry.mu.Unlock()
 169
 170         entry.latest = value
 171         entry.measurement = entry.measurement || value.measurement
 172         now := time.Now()
 173         equal := bytes.Equal(entry.successful, value.comparison)
 174         correlated := value.metadata.CorrelationID != "" || value.metadata.CausationID != ""
 175         due := entry.measurement && !now.Before(entry.retainedAt.Add(protocol.MeasurementRepeatInterval)) || value.online && !now.Before(entry.publishedAt.Add(protocol.OnlineRepeatInterval))
 176         if equal && !correlated && !due {
 177                 return nil
 178         }
 179
 180         if entry.measurement && equal && due && !correlated {
 181                 value, err = value.repeated()
 182                 if err != nil {
 183                         return fmt.Errorf("prepare reported repeat: %w", err)
 184                 }
 185
 186                 publication = value.publication
 187         }
 188
 189         err = p.send(ctx, publication)
 190         if err != nil {
 191                 return fmt.Errorf("send publication: %w", err)
 192         }
 193
 194         entry.successful = value.comparison
 195         entry.publishedAt = now
 196         if publication.Retain {
 197                 entry.retainedAt = now
 198         }
 199
 200         return nil
 201 }
 202
 203 func (p *Publisher) send(ctx context.Context, publication protocol.Publication) error {
 204         err := ctx.Err()
 205         if err != nil {
 206                 return err
 207         }
 208
 209         if publication.ExpiresAt != nil && !time.Now().Before(*publication.ExpiresAt) {
 210                 return fmt.Errorf("publication expired")
 211         }
 212
 213         ctx, cancel := context.WithTimeout(ctx, publishTimeout)
 214         defer cancel()
 215
 216         err = p.transport.Publish(ctx, publication)
 217         if err != nil {
 218                 return fmt.Errorf("send transport publication: %w", err)
 219         }
 220
 221         return nil
 222 }
 223
 224 func (p *Publisher) repeat(ctx context.Context, force bool) error {
 225         p.mu.Lock()
 226
 227         entries := make([]*retainedValue, 0, len(p.retained))
 228
 229         for _, entry := range p.retained {
 230                 entries = append(entries, entry)
 231         }
 232
 233         p.mu.Unlock()
 234
 235         for _, entry := range entries {
 236                 err := p.repeatEntry(ctx, entry, force)
 237                 if err != nil {
 238                         return fmt.Errorf("repeat retained value: %w", err)
 239                 }
 240         }
 241
 242         return nil
 243 }
 244
 245 func (p *Publisher) repeatEntry(ctx context.Context, entry *retainedValue, force bool) error {
 246         entry.mu.Lock()
 247         defer entry.mu.Unlock()
 248
 249         now := time.Now()
 250         due := entry.measurement && (!now.Before(entry.retainedAt.Add(protocol.MeasurementRepeatInterval)) || !bytes.Equal(entry.successful, entry.latest.comparison)) || entry.latest.online && !now.Before(entry.publishedAt.Add(protocol.OnlineRepeatInterval))
 251         if !force && !due {
 252                 return nil
 253         }
 254
 255         value, err := entry.latest.repeated()
 256         if err != nil {
 257                 return fmt.Errorf("prepare snapshot: %w", err)
 258         }
 259
 260         err = p.send(ctx, value.publication)
 261         if err != nil {
 262                 return fmt.Errorf("send snapshot: %w", err)
 263         }
 264
 265         entry.successful = value.comparison
 266         entry.publishedAt = now
 267         entry.retainedAt = now
 268
 269         return nil
 270 }
 271
 272 func publishTyped[T any](ctx context.Context, publisher *Publisher, address protocol.Address, metadata protocol.Metadata, payload T) error {
 273         value, err := newPublicationValue(address, metadata, payload)
 274         if err != nil {
 275                 return fmt.Errorf("prepare typed publication: %w", err)
 276         }
 277
 278         err = publisher.publish(ctx, value)
 279         if err != nil {
 280                 return fmt.Errorf("publish %s: %w", address.Kind, err)
 281         }
 282
 283         return nil
 284 }