Poll and command shared Alamo thermostats

[onehouse] / driver / m31 / driver.go

package m31

import (
        "context"
        "errors"
        "fmt"
        "net"
        "strconv"
        "time"

        "onehouse/driver"
        "onehouse/protocol"
)

// Driver serializes one assembly's polling and commands. Run has one owner and
// closes the supplied Client on each attempt. Recoverable device failures
// publish offline and retry with backoff. Publication or cleanup failures
// return to the framework runner. The shared publisher must run alongside this
// driver. Refresh is safe during Run. Duplicate protection ends at ledger
// eviction or process restart. No operation starts a physical connection before
// Run.
type Driver struct {
        deviceID     string
        address      string
        unit         byte
        pollInterval time.Duration
        topology     topology
        publisher    *driver.Publisher
        client       Client
        queue        <-chan []driver.Command
        ledger       *driver.Ledger
}

// New validates settings and requires the device's queue in router. The caller
// supplies valid publisher, router, and client dependencies and starts the
// router's intake separately. The driver owns the client until Run returns.
func New(settings Settings, publisher *driver.Publisher, router *driver.Router, client Client) (*Driver, error) {
        if !protocol.ValidID(settings.DeviceID) {
                return nil, fmt.Errorf("device ID must be a canonical ULID")
        }

        host, port, err := net.SplitHostPort(settings.Address)
        if err != nil || host == "" {
                return nil, fmt.Errorf("address must contain a host and TCP port")
        }

        number, err := strconv.Atoi(port)
        if err != nil || number < 1 || number > 65535 {
                return nil, fmt.Errorf("TCP port is outside its range")
        }

        if settings.Unit == 0 || settings.Unit > 247 {
                return nil, fmt.Errorf("unit must be from one through 247")
        }

        if settings.PollInterval < 10*time.Millisecond || settings.PollInterval > time.Minute {
                return nil, fmt.Errorf("poll interval must be from ten milliseconds through one minute")
        }

        points, err := newTopology(settings.Modules)
        if err != nil {
                return nil, fmt.Errorf("validate topology: %w", err)
        }

        queue, exists := router.Queue(settings.DeviceID)
        if !exists {
                return nil, fmt.Errorf("device has no command queue")
        }

        ledger, err := driver.NewLedger(settings.DeviceID)
        if err != nil {
                return nil, fmt.Errorf("create result ledger: %w", err)
        }

        return &Driver{
                deviceID:     settings.DeviceID,
                address:      settings.Address,
                unit:         settings.Unit,
                pollInterval: settings.PollInterval,
                topology:     points,
                publisher:    publisher,
                client:       client,
                queue:        queue,
                ledger:       ledger,
        }, nil
}

// Run configures and verifies output modes before online or command work. A
// successful first read establishes a baseline without input events. Device
// failures retry after one second, doubling to thirty seconds until a complete
// baseline succeeds. Commands received during backoff fail without I/O.
func (d *Driver) Run(ctx context.Context) error {
        err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
        if err != nil {
                return fmt.Errorf("publish descriptor: %w", err)
        }

        delay := time.Second

        for {
                healthy, retry, err := d.attempt(ctx)
                if !retry || ctx.Err() != nil {
                        return fmt.Errorf("run assembly: %w", err)
                }

                err = d.publisher.Availability(ctx, d.deviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"})
                if err != nil {
                        return fmt.Errorf("publish offline: %w", err)
                }

                if healthy {
                        delay = time.Second
                }

                err = d.wait(ctx, delay)
                if err != nil {
                        return fmt.Errorf("wait for reconnect: %w", err)
                }

                delay = min(delay*2, 30*time.Second)
        }
}

// Refresh requests retained snapshots through the shared publisher. It does not
// repeat input events, physical commands, or pulse operations.
func (d *Driver) Refresh() {
        d.publisher.Refresh()
}

func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err error) {
        defer func() {
                var failure *deviceError

                retry = errors.As(err, &failure)

                closeErr := d.client.Close()
                if closeErr != nil {
                        retry = false
                        err = fmt.Errorf("close assembly: %w", errors.Join(err, closeErr))
                }
        }()

        err = d.client.Connect(ctx, d.address, d.unit)
        if err != nil {
                return false, false, &deviceError{err: fmt.Errorf("connect assembly: %w", err)}
        }

        connection := &connection{owner: d}
        err = connection.start(ctx)
        if err != nil {
                return false, false, fmt.Errorf("initialize assembly: %w", err)
        }

        healthy = true
        handler := &commandHandler{connection: connection}
        ticker := time.NewTicker(d.pollInterval)
        defer ticker.Stop()

        for {
                select {
                case <-ctx.Done():
                        return healthy, false, ctx.Err()
                case commands := <-d.queue:
                        err = handler.handle(ctx, commands)
                        if err != nil {
                                return healthy, false, fmt.Errorf("process commands: %w", err)
                        }
                case <-ticker.C:
                        err = connection.poll(ctx)
                        if err != nil {
                                return healthy, false, fmt.Errorf("poll assembly: %w", err)
                        }
                }
        }
}

func (d *Driver) wait(ctx context.Context, delay time.Duration) error {
        timer := time.NewTimer(delay)
        defer timer.Stop()

        handler := &commandHandler{connection: &connection{owner: d}}

        for {
                select {
                case <-ctx.Done():
                        return ctx.Err()
                case <-timer.C:
                        return nil
                case commands := <-d.queue:
                        err := handler.unavailable(ctx, commands)
                        if err != nil {
                                return fmt.Errorf("finish unavailable commands: %w", err)
                        }
                }
        }
}