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

   1 package alamo
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "sync"
   8         "time"
   9
  10         "onehouse/driver"
  11         "onehouse/protocol"
  12 )
  13
  14 // Driver owns one gateway's shared polling, request budget, and command work.
  15 // Run has one owner. HTTP failures count toward missed polls or command failure
  16 // results. Publication failures return to the runner. No method logs. The
  17 // caller runs the shared publisher alongside the driver and owns the Client.
  18 // Refresh is safe during Run. Ledger protection is memory-only and ends at
  19 // eviction or process restart. Construct one driver per physical gateway.
  20 type Driver struct {
  21         gateway      gateway
  22         pollInterval time.Duration
  23         thermostats  []*thermostatState
  24         refresh      chan struct{}
  25         pending      chan queuedCommands
  26         publisher    *driver.Publisher
  27 }
  28
  29 // New validates settings and command queues without I/O. The caller supplies
  30 // valid publisher, router, and client dependencies, and starts router intake
  31 // separately. The constructor retains no caller-owned settings slice.
  32 func New(settings Settings, publisher *driver.Publisher, router *driver.Router, client Client) (*Driver, error) {
  33         if len(settings.Thermostats) == 0 {
  34                 return nil, fmt.Errorf("gateway requires at least one thermostat")
  35         }
  36
  37         interval := settings.PollInterval
  38         if interval == 0 {
  39                 interval = 15 * time.Second
  40         }
  41
  42         if interval < 10*time.Second || interval > 5*time.Minute {
  43                 return nil, fmt.Errorf("poll interval must be from ten seconds through five minutes")
  44         }
  45
  46         integration := &Driver{
  47                 gateway:      gateway{client: client},
  48                 pollInterval: interval,
  49                 refresh:      make(chan struct{}, 1),
  50                 pending:      make(chan queuedCommands),
  51                 publisher:    publisher,
  52         }
  53         ids := make(map[string]bool)
  54         macs := make(map[string]bool)
  55
  56         for _, thermostat := range settings.Thermostats {
  57                 if !protocol.ValidID(thermostat.DeviceID) || ids[thermostat.DeviceID] {
  58                         return nil, fmt.Errorf("thermostat device IDs must be unique canonical ULIDs")
  59                 }
  60
  61                 mac, err := normalizeMAC(thermostat.MAC)
  62                 if err != nil {
  63                         return nil, fmt.Errorf("validate thermostat MAC: %w", err)
  64                 }
  65
  66                 if macs[mac] {
  67                         return nil, fmt.Errorf("thermostat MAC values must be unique")
  68                 }
  69
  70                 queue, exists := router.Queue(thermostat.DeviceID)
  71                 if !exists {
  72                         return nil, fmt.Errorf("thermostat has no command queue")
  73                 }
  74
  75                 ledger, err := driver.NewLedger(thermostat.DeviceID)
  76                 if err != nil {
  77                         return nil, fmt.Errorf("create thermostat ledger: %w", err)
  78                 }
  79
  80                 ids[thermostat.DeviceID] = true
  81                 macs[mac] = true
  82                 thermostat.MAC = mac
  83                 integration.thermostats = append(integration.thermostats, &thermostatState{
  84                         settings:  thermostat,
  85                         queue:     queue,
  86                         ledger:    ledger,
  87                         publisher: publisher,
  88                 })
  89         }
  90
  91         return integration, nil
  92 }
  93
  94 // Run publishes descriptors, then shares each poll across the thermostat list.
  95 // Three missed observations mark a thermostat offline. A full observation
  96 // restores it. Shutdown does not publish offline. A command and its actions run
  97 // in order through the gateway budget and the shared result ledger.
  98 func (d *Driver) Run(ctx context.Context) error {
  99         ctx, cancel := context.WithCancel(ctx)
 100         defer cancel()
 101
 102         var workers sync.WaitGroup
 103
 104         defer func() {
 105                 cancel()
 106                 workers.Wait()
 107         }()
 108
 109         for _, thermostat := range d.thermostats {
 110                 err := thermostat.descriptor(ctx)
 111                 if err != nil {
 112                         return fmt.Errorf("publish thermostat descriptor: %w", err)
 113                 }
 114
 115                 workers.Go(func() {
 116                         d.forward(ctx, thermostat)
 117                 })
 118         }
 119
 120         err := d.poll(ctx)
 121         if err != nil {
 122                 return fmt.Errorf("poll initial thermostat states: %w", err)
 123         }
 124
 125         ticker := time.NewTicker(d.pollInterval)
 126         defer ticker.Stop()
 127
 128         for {
 129                 select {
 130                 case <-ctx.Done():
 131                         return ctx.Err()
 132                 case <-d.refresh:
 133                         err = d.poll(ctx)
 134                         if err != nil {
 135                                 return fmt.Errorf("refresh thermostat states: %w", err)
 136                         }
 137                 case <-ticker.C:
 138                         err = d.poll(ctx)
 139                         if err != nil {
 140                                 return fmt.Errorf("poll thermostat states: %w", err)
 141                         }
 142                 case pending := <-d.pending:
 143                         handler := commandHandler{
 144                                 gateway:    &d.gateway,
 145                                 thermostat: pending.thermostat,
 146                         }
 147
 148                         err = handler.handle(ctx, pending.commands)
 149                         if err != nil {
 150                                 return fmt.Errorf("process thermostat commands: %w", err)
 151                         }
 152                 }
 153         }
 154 }
 155
 156 func (d *Driver) forward(ctx context.Context, thermostat *thermostatState) {
 157         for {
 158                 select {
 159                 case <-ctx.Done():
 160                         return
 161                 case commands := <-thermostat.queue:
 162                         select {
 163                         case <-ctx.Done():
 164                                 return
 165                         case d.pending <- queuedCommands{
 166                                 thermostat: thermostat,
 167                                 commands:   commands,
 168                         }:
 169                         }
 170                 }
 171         }
 172 }
 173
 174 func (d *Driver) poll(ctx context.Context) error {
 175         states, err := d.gateway.status(ctx)
 176         if ctx.Err() != nil {
 177                 if err != nil {
 178                         return fmt.Errorf("stop gateway poll: %w", errors.Join(ctx.Err(), err))
 179                 }
 180
 181                 return ctx.Err()
 182         }
 183
 184         observedAt := time.Now().UTC()
 185         observed := make(map[string]State, len(states))
 186         if err == nil {
 187                 for _, state := range states {
 188                         observed[state.MAC] = state
 189                 }
 190         }
 191
 192         for _, thermostat := range d.thermostats {
 193                 state, exists := observed[thermostat.settings.MAC]
 194                 if exists {
 195                         err = thermostat.record(ctx, state, observedAt, "")
 196                 } else {
 197                         err = thermostat.miss(ctx)
 198                 }
 199
 200                 if err != nil {
 201                         return fmt.Errorf("record thermostat poll: %w", err)
 202                 }
 203         }
 204
 205         return nil
 206 }
 207
 208 // Refresh coalesces one rate-limited status poll and asks the shared publisher
 209 // to resend its snapshots. It does not repeat physical command actions.
 210 func (d *Driver) Refresh() {
 211         d.publisher.Refresh()
 212
 213         select {
 214         case d.refresh <- struct{}{}:
 215         default:
 216         }
 217 }