Poll and command shared Alamo thermostats

One gateway worker shares status polls and the request budget across its
thermostats. Three missed observations publish offline, and successful
feedback restores online. Commands read a fresh baseline, verify power
off before target changes, and verify the target before power on.

The shared ledger records each outcome before feedback or result sends,
so duplicate commands replay stable results without another exchange.
Caller cancellation preserves independent failures, and HTTP errors omit
sensitive request and response data. The command keeps its empty driver
list so construction cannot start a device connection.

Issue: 5947f38

diff --git a/README.rst b/README.rst
index 69b14d2f4b2d28da587b04cdc598a197a00b5eb3..55f1dab80949312a10136e1a8c428ab8c94d3996 100644 (file)
--- a/README.rst
+++ b/README.rst
@@ -5,7 +5,8 @@
 OneHouse is one program that controls a whole house. Linux is supported.
 The command runs an empty driver list and stops on SIGINT or SIGTERM.
 ``doc/driver.rst`` describes the framework and ``doc/mqtt.rst`` defines
 OneHouse is one program that controls a whole house. Linux is supported.
 The command runs an empty driver list and stops on SIGINT or SIGTERM.
 ``doc/driver.rst`` describes the framework and ``doc/mqtt.rst`` defines

-its wire contract. ``doc/m31.rst`` describes the M31 integration.
+its wire contract. ``doc/m31.rst`` and ``doc/alamo.rst`` describe the
+device integrations.

 
 *************
  Development
 
 *************
  Development

diff --git a/doc/alamo.rst b/doc/alamo.rst
index c9e6fde47ee62174b7eff135f25135036f3641c9..8c02dce69f91418d706e69dd2e66a3a06b8ece2c 100644 (file)
--- a/doc/alamo.rst
+++ b/doc/alamo.rst
@@ -33,3 +33,44 @@ The private control value uses ``settings.mode`` (``off`` or ``heat``),
 ``idle``, or ``heating``), and ``locked``. Commands contain the complete
 mode and target settings only. House-specific target bounds and steps
 belong to core validation. The driver accepts finite targets.
 ``idle``, or ``heating``), and ``locked``. Commands contain the complete
 mode and target settings only. House-specific target bounds and steps
 belong to core validation. The driver accepts finite targets.

+
+****************
+ Gateway driver
+****************
+
+``New`` takes ``Settings``, the shared publisher and router, and a raw
+``Client``. Settings list all stable thermostat IDs and their unique MAC
+values for one physical gateway. The default poll interval is fifteen
+seconds. An explicit interval is from ten seconds through five minutes.
+The caller runs the publisher, supplies command intake, and owns the
+HTTP transport. The command keeps its explicit empty driver list.
+
+``Run`` serializes polling and commands across the gateway. One status
+request supplies every configured thermostat. It publishes no startup
+availability before evidence. Three failed or missing observations mark
+a thermostat offline. A successful observation resets that count and
+restores online state. A command error does not count as a missed poll.
+Recoverable HTTP failures appear as missed polls or failed command
+results, not logs. Publication failures return to the framework runner.
+Shutdown does not publish an outage. ``Refresh`` coalesces a
+rate-limited status poll and asks the shared publisher to resend cached
+snapshots.
+
+Each command reads a fresh action baseline, turns off before a target
+change, sets and verifies the target, and only then turns on if
+requested. Every action response is feedback. A failed intermediate
+readback stops later actions. A command can leave the thermostat off if
+a later action fails or expires. The driver publishes the last
+successful observation with its observation time and command causation.
+A completed result requires matching power and target readback. A
+complete no-change command still reads the gateway.
+
+One queue consumer per thermostat preserves command order. Work from
+separate thermostats has no cross-device order guarantee. The shared
+ledger stores an outcome before feedback or result publication.
+Duplicate commands replay the first result without another gateway
+exchange, also after a failed publication. The ledger is memory-only
+with the framework limit of 256 results per device. Restart or eviction
+ends that duplicate protection. A gateway request can have an uncertain
+physical outcome when its response fails. The driver does not retry that
+command action.

diff --git a/driver/alamo/client.go b/driver/alamo/client.go
index de708463152d25d3583ccab72d35830af7a69803..29cfb41623b1fab750c080d3ec2640db2b9e4e41 100644 (file)
--- a/driver/alamo/client.go
+++ b/driver/alamo/client.go
@@ -5,7 +5,7 @@ import "context"
 // Client performs one raw gateway exchange. The driver serializes exchanges and
 // applies the shared request interval. Implementations honor ctx, return full
 // validated records, and do not log. The caller owns the client and its
 // Client performs one raw gateway exchange. The driver serializes exchanges and
 // applies the shared request interval. Implementations honor ctx, return full
 // validated records, and do not log. The caller owns the client and its

-// cleanup.
+// cleanup. Errors omit addresses, queries, and response data.

 type Client interface {
        Status(ctx context.Context) ([]State, error)
        Action(ctx context.Context, mac string, action Action) (State, error)
 type Client interface {
        Status(ctx context.Context) ([]State, error)
        Action(ctx context.Context, mac string, action Action) (State, error)

diff --git a/driver/alamo/command_handler.go b/driver/alamo/command_handler.go
new file mode 100644 (file)
index 0000000..535a4ba
--- /dev/null
+++ b/driver/alamo/command_handler.go
@@ -0,0 +1,165 @@
+package alamo
+
+import (
+       "context"
+       "errors"
+       "fmt"
+       "time"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+type commandHandler struct {
+       gateway    *gateway
+       thermostat *thermostatState
+}
+
+func (h *commandHandler) handle(ctx context.Context, commands []driver.Command) error {
+       for _, command := range commands {
+               err := h.one(ctx, command)
+               if err != nil {
+                       return fmt.Errorf("process thermostat command: %w", err)
+               }
+       }
+
+       return nil
+}
+
+func (h *commandHandler) one(ctx context.Context, command driver.Command) error {
+       s := h.thermostat
+       if command.DeviceID != s.settings.DeviceID {
+               return fmt.Errorf("command device is not owned by this thermostat")
+       }
+
+       record, exists := s.ledger.Lookup(command.Metadata.MessageID)
+       if exists {
+               err := s.publisher.Result(ctx, s.settings.DeviceID, command.Metadata.MessageID, record.Metadata, record.Payload)
+               if err != nil {
+                       return fmt.Errorf("replay thermostat result: %w", err)
+               }
+
+               return nil
+       }
+
+       reason := ""
+       settings, err := decodeControl(command.Payload.Value)
+
+       switch {
+       case command.Expired(time.Now()):
+               reason = "Command expired"
+       case command.Capability != "control":
+               reason = "Capability is not the thermostat control"
+       case err != nil:
+               reason = "Thermostat settings are invalid"
+       }
+
+       if reason != "" {
+               err = s.result(ctx, command, protocol.ResultPayload{
+                       Status: "rejected",
+                       Reason: reason,
+               })
+               if err != nil {
+                       return fmt.Errorf("reject thermostat command: %w", err)
+               }
+
+               return nil
+       }
+
+       state, observedAt, applyErr := h.apply(ctx, command, settings)
+       payload := protocol.ResultPayload{Status: "completed"}
+       if applyErr != nil {
+               payload.Status = "failed"
+               payload.Reason = "Thermostat request or readback failed"
+               if command.Expired(time.Now()) {
+                       payload.Reason = "Command expired"
+               }
+       }
+
+       record, err = s.ledger.Remember(command.Metadata.MessageID, payload)
+       if err != nil {
+               return fmt.Errorf("remember applied thermostat result: %w", err)
+       }
+
+       if ctx.Err() != nil {
+               if applyErr != nil {
+                       return fmt.Errorf("stop thermostat command: %w", errors.Join(ctx.Err(), applyErr))
+               }
+
+               return ctx.Err()
+       }
+
+       if !observedAt.IsZero() {
+               err = s.record(ctx, state, observedAt, command.Metadata.MessageID)
+               if err != nil {
+                       return fmt.Errorf("record command feedback: %w", err)
+               }
+       }
+
+       err = s.publisher.Result(ctx, s.settings.DeviceID, command.Metadata.MessageID, record.Metadata, record.Payload)
+       if err != nil {
+               return fmt.Errorf("publish applied thermostat result: %w", err)
+       }
+
+       return nil
+}
+
+func (h *commandHandler) apply(ctx context.Context, command driver.Command, settings controlSettings) (State, time.Time, error) {
+       ctx, cancel := context.WithDeadline(ctx, *command.Metadata.ExpiresAt)
+       defer cancel()
+
+       mac := h.thermostat.settings.MAC
+       state, err := h.gateway.action(ctx, mac, Action{})
+       if err != nil {
+               return State{}, time.Time{}, fmt.Errorf("read command baseline: %w", err)
+       }
+
+       observedAt := time.Now().UTC()
+
+       var next State
+
+       wantPower := settings.Mode == "heat"
+       if state.Power && (!wantPower || state.TargetTemperature != settings.TargetTemperature) {
+               power := false
+               next, err = h.gateway.action(ctx, mac, Action{Power: &power})
+               if err != nil {
+                       return state, observedAt, fmt.Errorf("turn thermostat off: %w", err)
+               }
+
+               state = next
+               observedAt = time.Now().UTC()
+               if state.Power {
+                       return state, observedAt, fmt.Errorf("thermostat remains on before target change")
+               }
+       }
+
+       if state.TargetTemperature != settings.TargetTemperature {
+               next, err = h.gateway.action(ctx, mac, Action{TargetTemperature: &settings.TargetTemperature})
+               if err != nil {
+                       return state, observedAt, fmt.Errorf("set thermostat target: %w", err)
+               }
+
+               state = next
+               observedAt = time.Now().UTC()
+               if state.TargetTemperature != settings.TargetTemperature {
+                       return state, observedAt, fmt.Errorf("thermostat target readback differs")
+               }
+       }
+
+       if wantPower && !state.Power {
+               power := true
+               next, err = h.gateway.action(ctx, mac, Action{Power: &power})
+               if err != nil {
+                       return state, observedAt, fmt.Errorf("turn thermostat on: %w", err)
+               }
+
+               state = next
+               observedAt = time.Now().UTC()
+       }
+
+       if state.Power != wantPower || state.TargetTemperature != settings.TargetTemperature {
+               return state, observedAt, fmt.Errorf("thermostat settings readback differs")
+       }
+
+       return state, observedAt, nil
+}

diff --git a/driver/alamo/command_handler_test.go b/driver/alamo/command_handler_test.go
new file mode 100644 (file)
index 0000000..8602a58
--- /dev/null
+++ b/driver/alamo/command_handler_test.go
@@ -0,0 +1,388 @@
+package alamo
+
+import (
+       "context"
+       "errors"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+func TestCommandHandlerOrdersPowerAndTargetWithActionReadback(t *testing.T) {
+       testCases := []struct {
+               name         string
+               initialPower bool
+               value        string
+               wantActions  []string
+       }{
+               {
+                       name:        "target precedes power on",
+                       value:       `{"mode":"heat","target_temperature":22}`,
+                       wantActions: []string{"read", "target", "on"},
+               },
+               {
+                       name:         "power off precedes target change",
+                       initialPower: true,
+                       value:        `{"mode":"off","target_temperature":22}`,
+                       wantActions:  []string{"read", "off", "target"},
+               },
+               {
+                       name:         "active target change turns off then on",
+                       initialPower: true,
+                       value:        `{"mode":"heat","target_temperature":22}`,
+                       wantActions:  []string{"read", "off", "target", "on"},
+               },
+               {
+                       name:        "equal settings still get fresh readback",
+                       value:       `{"mode":"off","target_temperature":20}`,
+                       wantActions: []string{"read"},
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               state := testState()
+                               state.Power = tc.initialPower
+                               client := &fakeGateway{states: []State{state}}
+                               transport := &fakeTransport{}
+                               integration, _ := testDriver(t, client, transport)
+                               handler := &commandHandler{
+                                       gateway:    &integration.gateway,
+                                       thermostat: integration.thermostats[0],
+                               }
+                               command := testCommand()
+                               command.Payload.Value = []byte(tc.value)
+                               require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
+                               require.Equal(t, tc.wantActions, actionNames(client.actions))
+                               require.Len(t, transport.messages(t, protocol.Result), 1)
+
+                               result := transport.messages(t, protocol.Result)[0]
+                               require.Equal(t, protocol.ResultPayload{Status: "completed"}, result.Payload)
+
+                               reported := transport.messages(t, protocol.Reported)
+                               require.Len(t, reported, 1)
+                               require.Equal(t, command.Metadata.MessageID, reported[0].Metadata.CausationID)
+
+                               for index := 1; index < len(client.starts); index++ {
+                                       require.GreaterOrEqual(t, client.starts[index].Sub(client.starts[index-1]), 6*time.Second)
+                               }
+                       })
+               })
+       }
+}
+
+func TestCommandHandlerStopsAtUnmatchedIntermediateReadback(t *testing.T) {
+       testCases := []struct {
+               name         string
+               power        bool
+               ignorePower  bool
+               ignoreTarget bool
+               wantActions  []string
+       }{
+               {
+                       name:        "failed off confirmation prevents target write",
+                       power:       true,
+                       ignorePower: true,
+                       wantActions: []string{"read", "off"},
+               },
+               {
+                       name:         "failed target confirmation prevents power on",
+                       ignoreTarget: true,
+                       wantActions:  []string{"read", "target"},
+               },
+               {
+                       name:        "failed power on confirmation does not complete",
+                       ignorePower: true,
+                       wantActions: []string{"read", "target", "on"},
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               state := testState()
+                               state.Power = tc.power
+                               client := &fakeGateway{
+                                       states:       []State{state},
+                                       ignorePower:  tc.ignorePower,
+                                       ignoreTarget: tc.ignoreTarget,
+                               }
+                               transport := &fakeTransport{}
+                               integration, _ := testDriver(t, client, transport)
+                               handler := commandHandler{
+                                       gateway:    &integration.gateway,
+                                       thermostat: integration.thermostats[0],
+                               }
+                               require.NoError(t, handler.handle(t.Context(), []driver.Command{testCommand()}))
+                               require.Equal(t, tc.wantActions, actionNames(client.actions))
+
+                               result, ok := transport.messages(t, protocol.Result)[0].Payload.(protocol.ResultPayload)
+                               require.True(t, ok)
+                               require.Equal(t, "failed", result.Status)
+                               require.NotEmpty(t, result.Reason)
+                       })
+               })
+       }
+}
+
+func TestCommandHandlerReplaysFirstResultAfterFeedbackOrResultSendFailure(t *testing.T) {
+       for _, kind := range []protocol.Kind{protocol.Reported, protocol.Result, protocol.Availability} {
+               t.Run(string(kind), func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               failure := errors.New("transport unavailable")
+                               transport := &fakeTransport{
+                                       failKind: kind,
+                                       failure:  failure,
+                               }
+                               client := &fakeGateway{states: []State{testState()}}
+                               integration, _ := testDriver(t, client, transport)
+                               handler := commandHandler{
+                                       gateway:    &integration.gateway,
+                                       thermostat: integration.thermostats[0],
+                               }
+                               command := testCommand()
+                               require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
+
+                               calls := len(client.actions)
+                               record, exists := integration.thermostats[0].ledger.Lookup(command.Metadata.MessageID)
+                               require.True(t, exists)
+                               require.Equal(t, "completed", record.Payload.Status)
+
+                               transport.failKind = ""
+
+                               require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
+                               require.Len(t, client.actions, calls)
+
+                               messages := transport.messages(t, protocol.Result)
+                               last := messages[len(messages)-1]
+                               require.Equal(t, record.Metadata, last.Metadata)
+                               require.Equal(t, record.Payload, last.Payload)
+                       })
+               })
+       }
+}
+
+func TestCommandHandlerRejectsInvalidCommandsWithoutGatewayWork(t *testing.T) {
+       testCases := []struct {
+               name   string
+               change func(*driver.Command)
+       }{
+               {
+                       name: "expired work is rejected",
+                       change: func(command *driver.Command) {
+                               expires := time.Now().Add(-time.Second).UTC()
+                               command.Metadata.ExpiresAt = &expires
+                       },
+               },
+               {
+                       name: "missing expiry is rejected",
+                       change: func(command *driver.Command) {
+                               command.Metadata.ExpiresAt = nil
+                       },
+               },
+               {
+                       name: "another capability is rejected",
+                       change: func(command *driver.Command) {
+                               command.Capability = "other"
+                       },
+               },
+               {
+                       name: "incomplete settings are rejected",
+                       change: func(command *driver.Command) {
+                               command.Payload.Value = []byte(`{}`)
+                       },
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       client := &fakeGateway{}
+                       transport := &fakeTransport{}
+                       integration, _ := testDriver(t, client, transport)
+                       handler := commandHandler{
+                               gateway:    &integration.gateway,
+                               thermostat: integration.thermostats[0],
+                       }
+                       command := testCommand()
+                       tc.change(&command)
+                       require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
+                       require.Empty(t, client.starts)
+
+                       messages := transport.messages(t, protocol.Result)
+                       require.Len(t, messages, 2)
+                       require.Equal(t, messages[0], messages[1])
+
+                       result, ok := messages[0].Payload.(protocol.ResultPayload)
+                       require.True(t, ok)
+                       require.Equal(t, "rejected", result.Status)
+               })
+       }
+}
+
+func TestCommandHandlerCommandExpiryStopsLaterActionsAndReplay(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               client := &fakeGateway{states: []State{testState()}}
+               transport := &fakeTransport{}
+               integration, _ := testDriver(t, client, transport)
+               handler := commandHandler{
+                       gateway:    &integration.gateway,
+                       thermostat: integration.thermostats[0],
+               }
+               command := testCommand()
+               expires := time.Now().Add(7 * time.Second).UTC()
+               command.Metadata.ExpiresAt = &expires
+               require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
+               require.Equal(t, []string{"read", "target"}, actionNames(client.actions))
+               require.False(t, client.states[0].Power)
+
+               reported := transport.messages(t, protocol.Reported)
+               require.Len(t, reported, 1)
+               require.Equal(t, client.starts[1].UTC(), *reported[0].Metadata.SentAt)
+               require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
+               require.Len(t, client.actions, 2)
+
+               messages := transport.messages(t, protocol.Result)
+               require.Len(t, messages, 2)
+               require.Equal(t, messages[0], messages[1])
+               require.Equal(t, protocol.ResultPayload{
+                       Status: "failed",
+                       Reason: "Command expired",
+               }, messages[0].Payload)
+       })
+}
+
+func TestCommandHandlerRetainsFailuresWithoutRepeatingPhysicalActions(t *testing.T) {
+       for _, position := range []int{1, 2, 3, 4} {
+               t.Run(actionFailureName(position), func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               state := testState()
+                               state.Power = true
+                               client := &fakeGateway{
+                                       states:       []State{state},
+                                       failActionAt: position,
+                               }
+                               transport := &fakeTransport{}
+                               integration, _ := testDriver(t, client, transport)
+                               handler := commandHandler{
+                                       gateway:    &integration.gateway,
+                                       thermostat: integration.thermostats[0],
+                               }
+                               command := testCommand()
+                               require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
+                               require.Len(t, client.actions, position)
+
+                               messages := transport.messages(t, protocol.Result)
+                               require.Len(t, messages, 2)
+                               require.Equal(t, messages[0], messages[1])
+
+                               payload, ok := messages[0].Payload.(protocol.ResultPayload)
+                               require.True(t, ok)
+                               require.Equal(t, "failed", payload.Status)
+                       })
+               })
+       }
+}
+
+func TestCommandHandlerValidatesOwnershipAndReturnsPublicationFailures(t *testing.T) {
+       client := &fakeGateway{}
+       failure := errors.New("result unavailable")
+       transport := &fakeTransport{
+               failKind: protocol.Result,
+               failure:  failure,
+       }
+       integration, _ := testDriver(t, client, transport)
+       handler := commandHandler{
+               gateway:    &integration.gateway,
+               thermostat: integration.thermostats[0],
+       }
+       command := testCommand()
+       command.DeviceID = testSecondID
+       require.ErrorContains(t, handler.handle(t.Context(), []driver.Command{command}), "not owned")
+
+       command = testCommand()
+       command.Capability = "invalid"
+       require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
+       require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
+       require.Empty(t, client.starts)
+}
+
+func TestCommandHandlerCancellationStopsBeforeAnotherActionAndKeepsOutcome(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               client := &fakeGateway{states: []State{testState()}}
+               transport := &fakeTransport{}
+               integration, _ := testDriver(t, client, transport)
+               handler := commandHandler{
+                       gateway:    &integration.gateway,
+                       thermostat: integration.thermostats[0],
+               }
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               done := make(chan error, 1)
+               command := testCommand()
+
+               go func() {
+                       done <- handler.handle(ctx, []driver.Command{command})
+               }()
+
+               synctest.Wait()
+               cancel()
+               require.ErrorIs(t, <-done, context.Canceled)
+               require.Len(t, client.actions, 1)
+
+               _, exists := integration.thermostats[0].ledger.Lookup(command.Metadata.MessageID)
+               require.True(t, exists)
+               require.Empty(t, transport.messages(t, protocol.Result))
+       })
+}
+
+func TestCommandHandlerKeepsIndependentFailureDuringCancellation(t *testing.T) {
+       failure := errors.New("independent gateway failure")
+       ctx, cancel := context.WithCancel(t.Context())
+       defer cancel()
+
+       client := &cancelingGateway{
+               cancel:  cancel,
+               failure: failure,
+       }
+       integration, _ := testDriver(t, client, &fakeTransport{})
+       handler := commandHandler{
+               gateway:    &integration.gateway,
+               thermostat: integration.thermostats[0],
+       }
+       err := handler.handle(ctx, []driver.Command{testCommand()})
+       require.ErrorIs(t, err, context.Canceled)
+       require.ErrorIs(t, err, failure)
+}
+
+func actionNames(actions []Action) []string {
+       var names []string
+
+       for _, action := range actions {
+               name := "read"
+               if action.TargetTemperature != nil {
+                       name = "target"
+               }
+
+               if action.Power != nil {
+                       name = "off"
+                       if *action.Power {
+                               name = "on"
+                       }
+               }
+
+               names = append(names, name)
+       }
+
+       return names
+}
+
+func actionFailureName(position int) string {
+       return []string{"", "failed initial read", "failed power off", "failed target write", "failed power on"}[position]
+}

diff --git a/driver/alamo/driver.go b/driver/alamo/driver.go
new file mode 100644 (file)
index 0000000..791594f
--- /dev/null
+++ b/driver/alamo/driver.go
@@ -0,0 +1,217 @@
+package alamo
+
+import (
+       "context"
+       "errors"
+       "fmt"
+       "sync"
+       "time"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+// Driver owns one gateway's shared polling, request budget, and command work.
+// Run has one owner. HTTP failures count toward missed polls or command failure
+// results. Publication failures return to the runner. No method logs. The
+// caller runs the shared publisher alongside the driver and owns the Client.
+// Refresh is safe during Run. Ledger protection is memory-only and ends at
+// eviction or process restart. Construct one driver per physical gateway.
+type Driver struct {
+       gateway      gateway
+       pollInterval time.Duration
+       thermostats  []*thermostatState
+       refresh      chan struct{}
+       pending      chan queuedCommands
+       publisher    *driver.Publisher
+}
+
+// New validates settings and command queues without I/O. The caller supplies
+// valid publisher, router, and client dependencies, and starts router intake
+// separately. The constructor retains no caller-owned settings slice.
+func New(settings Settings, publisher *driver.Publisher, router *driver.Router, client Client) (*Driver, error) {
+       if len(settings.Thermostats) == 0 {
+               return nil, fmt.Errorf("gateway requires at least one thermostat")
+       }
+
+       interval := settings.PollInterval
+       if interval == 0 {
+               interval = 15 * time.Second
+       }
+
+       if interval < 10*time.Second || interval > 5*time.Minute {
+               return nil, fmt.Errorf("poll interval must be from ten seconds through five minutes")
+       }
+
+       integration := &Driver{
+               gateway:      gateway{client: client},
+               pollInterval: interval,
+               refresh:      make(chan struct{}, 1),
+               pending:      make(chan queuedCommands),
+               publisher:    publisher,
+       }
+       ids := make(map[string]bool)
+       macs := make(map[string]bool)
+
+       for _, thermostat := range settings.Thermostats {
+               if !protocol.ValidID(thermostat.DeviceID) || ids[thermostat.DeviceID] {
+                       return nil, fmt.Errorf("thermostat device IDs must be unique canonical ULIDs")
+               }
+
+               mac, err := normalizeMAC(thermostat.MAC)
+               if err != nil {
+                       return nil, fmt.Errorf("validate thermostat MAC: %w", err)
+               }
+
+               if macs[mac] {
+                       return nil, fmt.Errorf("thermostat MAC values must be unique")
+               }
+
+               queue, exists := router.Queue(thermostat.DeviceID)
+               if !exists {
+                       return nil, fmt.Errorf("thermostat has no command queue")
+               }
+
+               ledger, err := driver.NewLedger(thermostat.DeviceID)
+               if err != nil {
+                       return nil, fmt.Errorf("create thermostat ledger: %w", err)
+               }
+
+               ids[thermostat.DeviceID] = true
+               macs[mac] = true
+               thermostat.MAC = mac
+               integration.thermostats = append(integration.thermostats, &thermostatState{
+                       settings:  thermostat,
+                       queue:     queue,
+                       ledger:    ledger,
+                       publisher: publisher,
+               })
+       }
+
+       return integration, nil
+}
+
+// Run publishes descriptors, then shares each poll across the thermostat list.
+// Three missed observations mark a thermostat offline. A full observation
+// restores it. Shutdown does not publish offline. A command and its actions run
+// in order through the gateway budget and the shared result ledger.
+func (d *Driver) Run(ctx context.Context) error {
+       ctx, cancel := context.WithCancel(ctx)
+       defer cancel()
+
+       var workers sync.WaitGroup
+
+       defer func() {
+               cancel()
+               workers.Wait()
+       }()
+
+       for _, thermostat := range d.thermostats {
+               err := thermostat.descriptor(ctx)
+               if err != nil {
+                       return fmt.Errorf("publish thermostat descriptor: %w", err)
+               }
+
+               workers.Go(func() {
+                       d.forward(ctx, thermostat)
+               })
+       }
+
+       err := d.poll(ctx)
+       if err != nil {
+               return fmt.Errorf("poll initial thermostat states: %w", err)
+       }
+
+       ticker := time.NewTicker(d.pollInterval)
+       defer ticker.Stop()
+
+       for {
+               select {
+               case <-ctx.Done():
+                       return ctx.Err()
+               case <-d.refresh:
+                       err = d.poll(ctx)
+                       if err != nil {
+                               return fmt.Errorf("refresh thermostat states: %w", err)
+                       }
+               case <-ticker.C:
+                       err = d.poll(ctx)
+                       if err != nil {
+                               return fmt.Errorf("poll thermostat states: %w", err)
+                       }
+               case pending := <-d.pending:
+                       handler := commandHandler{
+                               gateway:    &d.gateway,
+                               thermostat: pending.thermostat,
+                       }
+
+                       err = handler.handle(ctx, pending.commands)
+                       if err != nil {
+                               return fmt.Errorf("process thermostat commands: %w", err)
+                       }
+               }
+       }
+}
+
+func (d *Driver) forward(ctx context.Context, thermostat *thermostatState) {
+       for {
+               select {
+               case <-ctx.Done():
+                       return
+               case commands := <-thermostat.queue:
+                       select {
+                       case <-ctx.Done():
+                               return
+                       case d.pending <- queuedCommands{
+                               thermostat: thermostat,
+                               commands:   commands,
+                       }:
+                       }
+               }
+       }
+}
+
+func (d *Driver) poll(ctx context.Context) error {
+       states, err := d.gateway.status(ctx)
+       if ctx.Err() != nil {
+               if err != nil {
+                       return fmt.Errorf("stop gateway poll: %w", errors.Join(ctx.Err(), err))
+               }
+
+               return ctx.Err()
+       }
+
+       observedAt := time.Now().UTC()
+       observed := make(map[string]State, len(states))
+       if err == nil {
+               for _, state := range states {
+                       observed[state.MAC] = state
+               }
+       }
+
+       for _, thermostat := range d.thermostats {
+               state, exists := observed[thermostat.settings.MAC]
+               if exists {
+                       err = thermostat.record(ctx, state, observedAt, "")
+               } else {
+                       err = thermostat.miss(ctx)
+               }
+
+               if err != nil {
+                       return fmt.Errorf("record thermostat poll: %w", err)
+               }
+       }
+
+       return nil
+}
+
+// Refresh coalesces one rate-limited status poll and asks the shared publisher
+// to resend its snapshots. It does not repeat physical command actions.
+func (d *Driver) Refresh() {
+       d.publisher.Refresh()
+
+       select {
+       case d.refresh <- struct{}{}:
+       default:
+       }
+}

diff --git a/driver/alamo/driver_test.go b/driver/alamo/driver_test.go
new file mode 100644 (file)
index 0000000..fbeaebd
--- /dev/null
+++ b/driver/alamo/driver_test.go
@@ -0,0 +1,341 @@
+package alamo
+
+import (
+       "context"
+       "errors"
+       "strings"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+func TestDriverPollSharesOneSnapshotAcrossThermostats(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               first := testState()
+               second := first
+               second.MAC = "fc610e7f98ab"
+               client := &fakeGateway{states: []State{first, second}}
+               transport := &fakeTransport{}
+               integration, _ := testDriver(t, client, transport, Thermostat{
+                       DeviceID: testDeviceID,
+                       MAC:      testMAC,
+               }, Thermostat{
+                       DeviceID: testSecondID,
+                       MAC:      second.MAC,
+               })
+
+               err := integration.poll(t.Context())
+               require.NoError(t, err)
+               require.Len(t, client.starts, 1)
+               require.Len(t, transport.messages(t, protocol.Availability), 2)
+               require.Len(t, transport.messages(t, protocol.Reported), 2)
+       })
+}
+
+func TestDriverMarksOnlyMissingThermostatOfflineAfterThreePollsAndRecovers(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               second := testState()
+               second.MAC = "fc610e7f98ab"
+               client := &fakeGateway{states: []State{testState(), second}}
+               transport := &fakeTransport{}
+               integration, _ := testDriver(t, client, transport, Thermostat{
+                       DeviceID: testDeviceID,
+                       MAC:      testMAC,
+               }, Thermostat{
+                       DeviceID: testSecondID,
+                       MAC:      second.MAC,
+               })
+
+               require.NoError(t, integration.poll(t.Context()))
+
+               client.states = []State{second}
+
+               for range 2 {
+                       require.NoError(t, integration.poll(t.Context()))
+                       require.Len(t, transport.messages(t, protocol.Availability), 2)
+               }
+
+               require.NoError(t, integration.poll(t.Context()))
+
+               messages := transport.messages(t, protocol.Availability)
+               require.Len(t, messages, 3)
+               require.Equal(t, testDeviceID, messages[2].Address.ID)
+               require.Equal(t, protocol.AvailabilityPayload{Status: "offline"}, messages[2].Payload)
+               require.NoError(t, integration.poll(t.Context()))
+               require.Len(t, transport.messages(t, protocol.Availability), 3)
+
+               client.states = []State{testState(), second}
+
+               require.NoError(t, integration.poll(t.Context()))
+
+               messages = transport.messages(t, protocol.Availability)
+               require.Len(t, messages, 4)
+               require.Equal(t, protocol.AvailabilityPayload{Status: "online"}, messages[3].Payload)
+       })
+}
+
+func TestDriverFailedPollsHaveNoPrematureStartupAvailability(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               client := &fakeGateway{err: errors.New("gateway unavailable")}
+               transport := &fakeTransport{}
+               integration, _ := testDriver(t, client, transport)
+
+               for range 2 {
+                       require.NoError(t, integration.poll(t.Context()))
+                       require.Empty(t, transport.messages(t, protocol.Availability))
+               }
+
+               require.NoError(t, integration.poll(t.Context()))
+
+               messages := transport.messages(t, protocol.Availability)
+               require.Len(t, messages, 1)
+               require.Equal(t, protocol.AvailabilityPayload{Status: "offline"}, messages[0].Payload)
+
+               client.err = nil
+               client.states = []State{testState()}
+
+               require.NoError(t, integration.poll(t.Context()))
+
+               client.err = errors.New("gateway unavailable")
+
+               require.NoError(t, integration.poll(t.Context()))
+               require.Len(t, transport.messages(t, protocol.Availability), 2)
+       })
+}
+
+func TestDriverRunForwardsCommandsRefreshesAndStopsWithoutOutage(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               client := &fakeGateway{states: []State{testState()}}
+               transport := &fakeTransport{}
+               integration, router := testDriver(t, client, transport)
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               integration.Refresh()
+               integration.Refresh()
+
+               done := make(chan error, 1)
+
+               go func() {
+                       done <- integration.Run(ctx)
+               }()
+
+               synctest.Wait()
+               require.Len(t, transport.messages(t, protocol.Descriptor), 1)
+               require.Equal(t, 1, client.startCount())
+               time.Sleep(6 * time.Second)
+               synctest.Wait()
+               require.Equal(t, 2, client.startCount())
+
+               command := testCommand()
+               publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+               require.NoError(t, err)
+               require.NoError(t, router.Handle(ctx, publication.Topic, publication.Data, false))
+               require.NoError(t, router.Handle(ctx, publication.Topic, publication.Data, false))
+
+               time.Sleep(40 * time.Second)
+               synctest.Wait()
+               require.Len(t, transport.messages(t, protocol.Result), 2)
+               cancel()
+               require.ErrorIs(t, <-done, context.Canceled)
+               require.Len(t, client.actions, 3)
+
+               messages := transport.messages(t, protocol.Availability)
+
+               for _, message := range messages {
+                       require.Equal(t, protocol.AvailabilityPayload{Status: "online"}, message.Payload)
+               }
+       })
+}
+
+func TestDriverConstructorValidatesAndCopiesSettings(t *testing.T) {
+       router, err := driver.NewRouter([]string{testDeviceID, testSecondID})
+       require.NoError(t, err)
+
+       publisher := driver.NewPublisher(&fakeTransport{})
+       valid := Thermostat{
+               DeviceID: testDeviceID,
+               MAC:      testMAC,
+       }
+       testCases := []struct {
+               name     string
+               settings Settings
+       }{
+               {
+                       name: "empty gateway is rejected",
+               },
+               {
+                       name: "invalid ID is rejected",
+                       settings: Settings{Thermostats: []Thermostat{{
+                               DeviceID: "bad",
+                               MAC:      testMAC,
+                       }}},
+               },
+               {
+                       name: "invalid MAC is rejected",
+                       settings: Settings{Thermostats: []Thermostat{{
+                               DeviceID: testDeviceID,
+                               MAC:      "bad",
+                       }}},
+               },
+               {
+                       name:     "duplicate ID is rejected",
+                       settings: Settings{Thermostats: []Thermostat{valid, valid}},
+               },
+               {
+                       name: "MAC uniqueness ignores letter case",
+                       settings: Settings{Thermostats: []Thermostat{valid, {
+                               DeviceID: testSecondID,
+                               MAC:      strings.ToUpper(testMAC),
+                       }}},
+               },
+               {
+                       name: "poll interval cannot exceed five minutes",
+                       settings: Settings{
+                               PollInterval: time.Hour,
+                               Thermostats:  []Thermostat{valid},
+                       },
+               },
+               {
+                       name: "poll interval cannot be below ten seconds",
+                       settings: Settings{
+                               PollInterval: time.Second,
+                               Thermostats:  []Thermostat{valid},
+                       },
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       _, err = New(tc.settings, publisher, router, &fakeGateway{})
+                       require.Error(t, err)
+               })
+       }
+
+       emptyRouter, err := driver.NewRouter(nil)
+       require.NoError(t, err)
+
+       _, err = New(Settings{Thermostats: []Thermostat{valid}}, publisher, emptyRouter, &fakeGateway{})
+       require.ErrorContains(t, err, "queue")
+
+       settings := Settings{
+               PollInterval: 10 * time.Second,
+               Thermostats:  []Thermostat{valid},
+       }
+       integration, err := New(settings, publisher, router, &fakeGateway{})
+       require.NoError(t, err)
+
+       settings.Thermostats[0].MAC = "bad"
+       require.Equal(t, testMAC, integration.thermostats[0].settings.MAC)
+       require.Equal(t, 10*time.Second, integration.pollInterval)
+}
+
+func TestDriverReturnsPublicationFailuresAndCallerCancellation(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               failure := errors.New("publication unavailable")
+               transport := &fakeTransport{
+                       failKind: protocol.Descriptor,
+                       failure:  failure,
+               }
+               integration, _ := testDriver(t, &fakeGateway{states: []State{testState()}}, transport)
+               require.ErrorIs(t, integration.Run(t.Context()), failure)
+
+               transport.failKind = protocol.Reported
+
+               require.ErrorIs(t, integration.Run(t.Context()), failure)
+
+               transport.failKind = protocol.Availability
+               integration.thermostats[0].missed = 2
+               integration.gateway.client = &fakeGateway{}
+               require.ErrorIs(t, integration.poll(t.Context()), failure)
+
+               ctx, cancel := context.WithCancel(t.Context())
+               cancel()
+
+               require.ErrorIs(t, integration.poll(ctx), context.Canceled)
+       })
+}
+
+func TestDriverReturnsRefreshAndPeriodicPublicationFailures(t *testing.T) {
+       for _, refresh := range []bool{false, true} {
+               t.Run(publicationFailureName(refresh), func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               client := &fakeGateway{states: []State{testState()}}
+                               transport := &fakeTransport{}
+                               integration, _ := testDriver(t, client, transport)
+                               ctx, cancel := context.WithCancel(t.Context())
+                               defer cancel()
+
+                               done := make(chan error, 1)
+
+                               go func() {
+                                       done <- integration.Run(ctx)
+                               }()
+
+                               synctest.Wait()
+
+                               failure := errors.New("publication unavailable")
+
+                               transport.mu.Lock()
+
+                               transport.failKind = protocol.Reported
+                               transport.failure = failure
+                               transport.mu.Unlock()
+                               client.mu.Lock()
+
+                               client.states[0].TargetTemperature = 25
+                               client.mu.Unlock()
+
+                               if refresh {
+                                       integration.Refresh()
+                               }
+
+                               require.ErrorIs(t, <-done, failure)
+                       })
+               })
+       }
+}
+
+func TestDriverReturnsCommandPublicationFailuresAndJoinsForwarders(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               client := &fakeGateway{states: []State{testState()}}
+               transport := &fakeTransport{
+                       failKind: protocol.Result,
+                       failure:  errors.New("result unavailable"),
+               }
+               integration, router := testDriver(t, client, transport)
+               command := testCommand()
+               publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+               require.NoError(t, err)
+               require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, false))
+               require.ErrorIs(t, integration.Run(t.Context()), transport.failure)
+       })
+}
+
+func TestDriverPollKeepsIndependentFailureDuringCancellation(t *testing.T) {
+       failure := errors.New("independent gateway failure")
+       ctx, cancel := context.WithCancel(t.Context())
+       defer cancel()
+
+       client := &cancelingGateway{
+               cancel:  cancel,
+               failure: failure,
+       }
+       integration, _ := testDriver(t, client, &fakeTransport{})
+       err := integration.poll(ctx)
+       require.ErrorIs(t, err, context.Canceled)
+       require.ErrorIs(t, err, failure)
+}
+
+func publicationFailureName(refresh bool) string {
+       if refresh {
+               return "refresh failure returns to runner"
+       }
+
+       return "periodic failure returns to runner"
+}

diff --git a/driver/alamo/fake_helper_test.go b/driver/alamo/fake_helper_test.go
index 03d36f2a2132d179ad197782b858c8518d49b2fb..bef18e28ac1fe7e15d9c681c7d103dfc42efa8fd 100644 (file)
--- a/driver/alamo/fake_helper_test.go
+++ b/driver/alamo/fake_helper_test.go
@@ -2,20 +2,43 @@ package alamo
 
 import (
        "context"
 
 import (
        "context"

+       "errors"
+       "fmt"
+       "sync"
+       "testing"

        "time"
        "time"

+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/driver"
+       "onehouse/protocol"

 )
 
 )
 

-const testMAC = "fc610e7f98aa"
+const (
+       validRecord = `{"mac":"FC610E7F98AA","power":0,"stemp":20,"temp":19.5,"sensor":23,"ativo":0,"bloqueio":1}`
+
+       testMAC       = "fc610e7f98aa"
+       testDeviceID  = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+       testSecondID  = "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+       testCommandID = "01ARZ3NDEKTSV4RRFFQ69G5FAX"
+)

 
 type fakeGateway struct {
 
 type fakeGateway struct {

-       states  []State
-       starts  []time.Time
-       actions []Action
-       err     error
-       delay   time.Duration
+       mu           sync.Mutex
+       failActionAt int
+       ignorePower  bool
+       ignoreTarget bool
+       states       []State
+       starts       []time.Time
+       actions      []Action
+       err          error
+       delay        time.Duration

 }
 
 func (g *fakeGateway) Status(ctx context.Context) ([]State, error) {
 }
 
 func (g *fakeGateway) Status(ctx context.Context) ([]State, error) {

+       g.mu.Lock()
+       defer g.mu.Unlock()
+

        err := g.request(ctx)
        if err != nil {
                return nil, err
        err := g.request(ctx)
        if err != nil {
                return nil, err


@@ -24,13 +47,36 @@ func (g *fakeGateway) Status(ctx context.Context) ([]State, error) {
        return append([]State(nil), g.states...), nil
 }
 
        return append([]State(nil), g.states...), nil
 }
 

+func (g *fakeGateway) request(ctx context.Context) error {
+       g.starts = append(g.starts, time.Now())
+       if g.delay > 0 {
+               timer := time.NewTimer(g.delay)
+               defer timer.Stop()
+
+               select {
+               case <-ctx.Done():
+                       return ctx.Err()
+               case <-timer.C:
+               }
+       }
+
+       return g.err
+}
+

 func (g *fakeGateway) Action(ctx context.Context, mac string, action Action) (State, error) {
 func (g *fakeGateway) Action(ctx context.Context, mac string, action Action) (State, error) {

+       g.mu.Lock()
+       defer g.mu.Unlock()
+
+       g.actions = append(g.actions, action)
+

        err := g.request(ctx)
        if err != nil {
                return State{}, err
        }
 
        err := g.request(ctx)
        if err != nil {
                return State{}, err
        }
 

-       g.actions = append(g.actions, action)
+       if len(g.actions) == g.failActionAt {
+               return State{}, errors.New("action unavailable")
+       }

 
        for index := range g.states {
                state := &g.states[index]
 
        for index := range g.states {
                state := &g.states[index]


@@ -38,11 +84,11 @@ func (g *fakeGateway) Action(ctx context.Context, mac string, action Action) (St
                        continue
                }
 
                        continue
                }
 

-               if action.Power != nil {
+               if action.Power != nil && !g.ignorePower {

                        state.Power = *action.Power
                }
 
                        state.Power = *action.Power
                }
 

-               if action.TargetTemperature != nil {
+               if action.TargetTemperature != nil && !g.ignoreTarget {

                        state.TargetTemperature = *action.TargetTemperature
                }
 
                        state.TargetTemperature = *action.TargetTemperature
                }
 


@@ -52,20 +98,110 @@ func (g *fakeGateway) Action(ctx context.Context, mac string, action Action) (St
        return State{}, nil
 }
 
        return State{}, nil
 }
 

-func (g *fakeGateway) request(ctx context.Context) error {
-       g.starts = append(g.starts, time.Now())
-       if g.delay > 0 {
-               timer := time.NewTimer(g.delay)
-               defer timer.Stop()
+func (g *fakeGateway) startCount() int {
+       g.mu.Lock()
+       defer g.mu.Unlock()

 
 

-               select {
-               case <-ctx.Done():
-                       return ctx.Err()
-               case <-timer.C:
+       return len(g.starts)
+}
+
+type fakeTransport struct {
+       mu           sync.Mutex
+       publications []protocol.Publication
+       failKind     protocol.Kind
+       failure      error
+}
+
+func (p *fakeTransport) Publish(_ context.Context, publication protocol.Publication) error {
+       p.mu.Lock()
+       defer p.mu.Unlock()
+
+       p.publications = append(p.publications, publication)
+       message, err := protocol.Parse(publication.Topic, publication.Data, publication.Retain)
+       if err != nil {
+               return fmt.Errorf("parse fake publication: %w", err)
+       }
+
+       if message.Address.Kind == p.failKind {
+               return p.failure
+       }
+
+       return nil
+}
+
+func (p *fakeTransport) messages(t *testing.T, kind protocol.Kind) []protocol.Message[any] {
+       t.Helper()
+       p.mu.Lock()
+       defer p.mu.Unlock()
+
+       var messages []protocol.Message[any]
+
+       for _, publication := range p.publications {
+               message, err := protocol.Parse(publication.Topic, publication.Data, publication.Retain)
+               require.NoError(t, err)
+
+               if message.Address.Kind == kind {
+                       messages = append(messages, message)

                }
        }
 
                }
        }
 

-       return g.err
+       return messages
+}
+
+type cancelingGateway struct {
+       cancel  context.CancelFunc
+       failure error
+}
+
+func (g *cancelingGateway) Status(_ context.Context) ([]State, error) {
+       g.cancel()
+
+       return nil, errors.Join(context.Canceled, g.failure)
+}
+
+func (g *cancelingGateway) Action(_ context.Context, _ string, _ Action) (State, error) {
+       g.cancel()
+
+       return State{}, errors.Join(context.Canceled, g.failure)
+}
+
+func testDriver(t *testing.T, client Client, transport *fakeTransport, thermostats ...Thermostat) (*Driver, *driver.Router) {
+       t.Helper()
+
+       if len(thermostats) == 0 {
+               thermostats = []Thermostat{{
+                       DeviceID: testDeviceID,
+                       MAC:      testMAC,
+               }}
+       }
+
+       var ids []string
+
+       for _, thermostat := range thermostats {
+               ids = append(ids, thermostat.DeviceID)
+       }
+
+       router, err := driver.NewRouter(ids)
+       require.NoError(t, err)
+
+       integration, err := New(Settings{Thermostats: thermostats}, driver.NewPublisher(transport), router, client)
+       require.NoError(t, err)
+
+       return integration, router
+}
+
+func testCommand() driver.Command {
+       expires := time.Now().Add(time.Minute).UTC()
+
+       return driver.Command{
+               DeviceID:   testDeviceID,
+               Capability: "control",
+               Metadata: protocol.Metadata{
+                       MessageID: testCommandID,
+                       ExpiresAt: &expires,
+               },
+               Payload: protocol.ValuePayload{Value: []byte(`{"mode":"heat","target_temperature":22}`)},
+       }

 }
 
 func testState() State {
 }
 
 func testState() State {

diff --git a/driver/alamo/gateway.go b/driver/alamo/gateway.go
index 1bf0074beb154fee54a1d51927f54c18ae33deed..2f3da3bcf96daa76ca8baa83437844c84f232f62 100644 (file)
--- a/driver/alamo/gateway.go
+++ b/driver/alamo/gateway.go
@@ -33,23 +33,6 @@ func (g *gateway) status(ctx context.Context) ([]State, error) {
        return states, nil
 }
 
        return states, nil
 }
 

-func (g *gateway) action(ctx context.Context, mac string, action Action) (State, error) {
-       err := g.wait(ctx)
-       if err != nil {
-               return State{}, fmt.Errorf("wait for action slot: %w", err)
-       }
-
-       ctx, cancel := context.WithTimeout(ctx, gatewayExchangeTimeout)
-       defer cancel()
-
-       state, err := g.client.Action(ctx, mac, action)
-       if err != nil {
-               return State{}, fmt.Errorf("exchange gateway action: %w", err)
-       }
-
-       return state, nil
-}
-

 func (g *gateway) wait(ctx context.Context) error {
        err := ctx.Err()
        if err != nil {
 func (g *gateway) wait(ctx context.Context) error {
        err := ctx.Err()
        if err != nil {


@@ -73,7 +56,29 @@ func (g *gateway) wait(ctx context.Context) error {
                return err
        }
 
                return err
        }
 

+       deadline, exists := ctx.Deadline()
+       if exists && !time.Now().Before(deadline) {
+               return context.DeadlineExceeded
+       }
+

        g.nextStart = time.Now().Add(gatewayRequestInterval)
 
        return nil
 }
        g.nextStart = time.Now().Add(gatewayRequestInterval)
 
        return nil
 }

+
+func (g *gateway) action(ctx context.Context, mac string, action Action) (State, error) {
+       err := g.wait(ctx)
+       if err != nil {
+               return State{}, fmt.Errorf("wait for action slot: %w", err)
+       }
+
+       ctx, cancel := context.WithTimeout(ctx, gatewayExchangeTimeout)
+       defer cancel()
+
+       state, err := g.client.Action(ctx, mac, action)
+       if err != nil {
+               return State{}, fmt.Errorf("exchange gateway action: %w", err)
+       }
+
+       return state, nil
+}

diff --git a/driver/alamo/gateway_record_test.go b/driver/alamo/gateway_record_test.go
new file mode 100644 (file)
index 0000000..529dcdd
--- /dev/null
+++ b/driver/alamo/gateway_record_test.go
@@ -0,0 +1,27 @@
+package alamo
+
+import (
+       "encoding/json"
+       "math"
+       "testing"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestGatewayRecordRejectsInvalidMACAndNonfiniteTemperatures(t *testing.T) {
+       var record gatewayRecord
+
+       require.NoError(t, json.Unmarshal([]byte(validRecord), &record))
+
+       invalid := "zzzzzzzzzzzz"
+       record.MAC = &invalid
+       _, err := record.state()
+       require.Error(t, err)
+
+       valid := testMAC
+       record.MAC = &valid
+       nonfinite := math.NaN()
+       record.Floor = &nonfinite
+       _, err = record.state()
+       require.ErrorContains(t, err, "finite")
+}

diff --git a/driver/alamo/gateway_test.go b/driver/alamo/gateway_test.go
index 715e860d41e49f70f4926fe09d4bdb578b0793e5..ab004e0d3570a6724932d0d5875c2dbbcc900689 100644 (file)
--- a/driver/alamo/gateway_test.go
+++ b/driver/alamo/gateway_test.go
@@ -85,3 +85,19 @@ func TestGatewayBoundsFakeClientOperationsAndReturnsFailures(t *testing.T) {
                require.ErrorIs(t, err, client.err)
        })
 }
                require.ErrorIs(t, err, client.err)
        })
 }

+
+func TestGatewayDoesNotStartAtTheExactCommandDeadline(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               client := &fakeGateway{states: []State{testState()}}
+               gateway := &gateway{client: client}
+               _, err := gateway.status(t.Context())
+               require.NoError(t, err)
+
+               ctx, cancel := context.WithTimeout(t.Context(), 6*time.Second)
+               defer cancel()
+
+               _, err = gateway.action(ctx, testMAC, Action{})
+               require.ErrorIs(t, err, context.DeadlineExceeded)
+               require.Len(t, client.starts, 1)
+       })
+}

diff --git a/driver/alamo/http_client.go b/driver/alamo/http_client.go
index 717912402c8f306cde39ac2031f5c66755ae06dd..4c645f405be125b452e20688069b920f17582118 100644 (file)
--- a/driver/alamo/http_client.go
+++ b/driver/alamo/http_client.go
@@ -54,47 +54,6 @@ func (c *HTTPClient) Status(ctx context.Context) ([]State, error) {
        return c.exchange(ctx, "/status", nil)
 }
 
        return c.exchange(ctx, "/status", nil)
 }
 

-// Action changes one field or reads a thermostat, and returns its complete
-// readback. It rejects a response for another MAC or more than one thermostat.
-func (c *HTTPClient) Action(ctx context.Context, mac string, action Action) (State, error) {
-       mac, err := normalizeMAC(mac)
-       if err != nil {
-               return State{}, fmt.Errorf("validate action MAC: %w", err)
-       }
-
-       if action.Power != nil && action.TargetTemperature != nil {
-               return State{}, fmt.Errorf("action must change at most one field")
-       }
-
-       query := url.Values{"mac": {mac}}
-       if action.Power != nil {
-               query.Set("power", "0")
-
-               if *action.Power {
-                       query.Set("power", "1")
-               }
-       }
-
-       if action.TargetTemperature != nil {
-               if !finite(*action.TargetTemperature) {
-                       return State{}, fmt.Errorf("target temperature must be finite")
-               }
-
-               query.Set("stemp", strconv.FormatFloat(*action.TargetTemperature, 'f', -1, 64))
-       }
-
-       states, err := c.exchange(ctx, "/acao", query)
-       if err != nil {
-               return State{}, fmt.Errorf("exchange action: %w", err)
-       }
-
-       if len(states) != 1 || states[0].MAC != mac {
-               return State{}, fmt.Errorf("action response must contain the requested thermostat only")
-       }
-
-       return states[0], nil
-}
-

 func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values) (states []State, err error) {
        ctx, cancel := context.WithTimeout(ctx, httpRequestTimeout)
        defer cancel()
 func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values) (states []State, err error) {
        ctx, cancel := context.WithTimeout(ctx, httpRequestTimeout)
        defer cancel()


@@ -111,11 +70,7 @@ func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values
 
        response, err := c.client.Do(request)
        if err != nil {
 
        response, err := c.client.Do(request)
        if err != nil {

-               if ctx.Err() != nil {
-                       return nil, ctx.Err()
-               }
-
-               return nil, fmt.Errorf("send gateway request")
+               return nil, fmt.Errorf("send gateway request: %w", safeRequestError(err))

        }
 
        defer func() {
        }
 
        defer func() {


@@ -131,11 +86,7 @@ func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values
 
        body, err := io.ReadAll(io.LimitReader(response.Body, gatewayResponseLimit+1))
        if err != nil {
 
        body, err := io.ReadAll(io.LimitReader(response.Body, gatewayResponseLimit+1))
        if err != nil {

-               if ctx.Err() != nil {
-                       return nil, ctx.Err()
-               }
-
-               return nil, fmt.Errorf("read gateway response")
+               return nil, fmt.Errorf("read gateway response: %w", safeRequestError(err))

        }
 
        if len(body) > gatewayResponseLimit {
        }
 
        if len(body) > gatewayResponseLimit {


@@ -152,9 +103,11 @@ func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values
        seen := make(map[string]bool, len(records))
 
        for _, record := range records {
        seen := make(map[string]bool, len(records))
 
        for _, record := range records {

-               state, recordErr := record.state()
-               if recordErr != nil {
-                       return nil, fmt.Errorf("validate gateway record: %w", recordErr)
+               var state State
+
+               state, err = record.state()
+               if err != nil {
+                       return nil, fmt.Errorf("validate gateway record: %w", err)

                }
 
                if seen[state.MAC] {
                }
 
                if seen[state.MAC] {


@@ -167,3 +120,70 @@ func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values
 
        return states, nil
 }
 
        return states, nil
 }

+
+// Action changes one field or reads a thermostat, and returns its complete
+// readback. It rejects a response for another MAC or more than one thermostat.
+func (c *HTTPClient) Action(ctx context.Context, mac string, action Action) (State, error) {
+       mac, err := normalizeMAC(mac)
+       if err != nil {
+               return State{}, fmt.Errorf("validate action MAC: %w", err)
+       }
+
+       if action.Power != nil && action.TargetTemperature != nil {
+               return State{}, fmt.Errorf("action must change at most one field")
+       }
+
+       query := url.Values{"mac": {mac}}
+       if action.Power != nil {
+               query.Set("power", "0")
+
+               if *action.Power {
+                       query.Set("power", "1")
+               }
+       }
+
+       if action.TargetTemperature != nil {
+               if !finite(*action.TargetTemperature) {
+                       return State{}, fmt.Errorf("target temperature must be finite")
+               }
+
+               query.Set("stemp", strconv.FormatFloat(*action.TargetTemperature, 'f', -1, 64))
+       }
+
+       states, err := c.exchange(ctx, "/acao", query)
+       if err != nil {
+               return State{}, fmt.Errorf("exchange action: %w", err)
+       }
+
+       if len(states) != 1 || states[0].MAC != mac {
+               return State{}, fmt.Errorf("action response must contain the requested thermostat only")
+       }
+
+       return states[0], nil
+}
+
+func safeRequestError(err error) error {
+       var joined interface{ Unwrap() []error }
+
+       if errors.As(err, &joined) {
+               var causes []error
+
+               for _, cause := range joined.Unwrap() {
+                       if cause != nil {
+                               causes = append(causes, safeRequestError(cause))
+                       }
+               }
+
+               return errors.Join(causes...)
+       }
+
+       if errors.Is(err, context.Canceled) {
+               return context.Canceled
+       }
+
+       if errors.Is(err, context.DeadlineExceeded) {
+               return context.DeadlineExceeded
+       }
+
+       return fmt.Errorf("gateway transport failed")
+}

diff --git a/driver/alamo/http_client_test.go b/driver/alamo/http_client_test.go
index 7ee0e36675e8bb71bba2d8d8236a2b97a1c6b5d3..1c8f8a01b5a1e4495e1f2d7f2ea7f40a6198abbc 100644 (file)
--- a/driver/alamo/http_client_test.go
+++ b/driver/alamo/http_client_test.go
@@ -15,8 +15,6 @@ import (
        "github.com/stretchr/testify/require"
 )
 
        "github.com/stretchr/testify/require"
 )
 

-const validRecord = `{"mac":"FC610E7F98AA","power":0,"stemp":20,"temp":19.5,"sensor":23,"ativo":0,"bloqueio":1}`
-

 func TestHTTPClientReadsAllRecordsAndOrderedActionFields(t *testing.T) {
        var requests []*http.Request
 
 func TestHTTPClientReadsAllRecordsAndOrderedActionFields(t *testing.T) {
        var requests []*http.Request
 


@@ -251,6 +249,18 @@ func TestHTTPClientDoesNotFollowRedirects(t *testing.T) {
        require.Equal(t, 1, calls)
 }
 
        require.Equal(t, 1, calls)
 }
 

+func TestHTTPClientKeepsIndependentFailureAlongsideCancellation(t *testing.T) {
+       client, err := NewHTTPClient("http://example.com", &http.Client{Transport: roundTripFunc(func(_ *http.Request) (*http.Response, error) {
+               return nil, errors.Join(context.Canceled, errors.New("secret"))
+       })})
+       require.NoError(t, err)
+
+       _, err = client.Status(t.Context())
+       require.ErrorIs(t, err, context.Canceled)
+       require.ErrorContains(t, err, "gateway transport failed")
+       require.NotContains(t, err.Error(), "secret")
+}
+

 type roundTripFunc func(*http.Request) (*http.Response, error)
 
 func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
 type roundTripFunc func(*http.Request) (*http.Response, error)
 
 func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {

diff --git a/driver/alamo/queued_commands.go b/driver/alamo/queued_commands.go
new file mode 100644 (file)
index 0000000..6d400f9
--- /dev/null
+++ b/driver/alamo/queued_commands.go
@@ -0,0 +1,8 @@
+package alamo
+
+import "onehouse/driver"
+
+type queuedCommands struct {
+       thermostat *thermostatState
+       commands   []driver.Command
+}

diff --git a/driver/alamo/settings.go b/driver/alamo/settings.go
new file mode 100644 (file)
index 0000000..20476df
--- /dev/null
+++ b/driver/alamo/settings.go
@@ -0,0 +1,12 @@
+package alamo
+
+import "time"
+
+// Settings defines one gateway's complete thermostat list. PollInterval
+// defaults to fifteen seconds when zero, and otherwise is from ten seconds
+// through five minutes. Use exactly one driver and client owner per physical
+// gateway. The constructor copies the list and performs no I/O.
+type Settings struct {
+       PollInterval time.Duration
+       Thermostats  []Thermostat
+}

diff --git a/driver/alamo/thermostat.go b/driver/alamo/thermostat.go
new file mode 100644 (file)
index 0000000..2cc908a
--- /dev/null
+++ b/driver/alamo/thermostat.go
@@ -0,0 +1,9 @@
+package alamo
+
+// Thermostat maps one stable canonical device ID to one gateway MAC. MAC
+// contains twelve hexadecimal digits and is normalized to lowercase. IDs and
+// MAC values must be unique in the gateway list.
+type Thermostat struct {
+       DeviceID string
+       MAC      string
+}

diff --git a/driver/alamo/thermostat_state.go b/driver/alamo/thermostat_state.go
new file mode 100644 (file)
index 0000000..f2fc524
--- /dev/null
+++ b/driver/alamo/thermostat_state.go
@@ -0,0 +1,91 @@
+package alamo
+
+import (
+       "context"
+       "encoding/json"
+       "fmt"
+       "time"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+type thermostatState struct {
+       settings  Thermostat
+       queue     <-chan []driver.Command
+       ledger    *driver.Ledger
+       publisher *driver.Publisher
+       missed    int
+}
+
+func (s *thermostatState) descriptor(ctx context.Context) error {
+       err := s.publisher.Descriptor(ctx, s.settings.DeviceID, protocol.Metadata{}, protocol.DescriptorPayload{
+               Manufacturer: "alamo",
+               Model:        "slim-wifi",
+               Firmware:     "unknown",
+               Capabilities: []protocol.Capability{{
+                       ID:              "control",
+                       Type:            "floor-heating",
+                       ReportsState:    true,
+                       AcceptsCommands: true,
+               }},
+       })
+       if err != nil {
+               return fmt.Errorf("publish descriptor: %w", err)
+       }
+
+       return nil
+}
+
+func (s *thermostatState) record(ctx context.Context, state State, observedAt time.Time, causationID string) error {
+       s.missed = 0
+       metadata := protocol.Metadata{
+               SentAt:      &observedAt,
+               CausationID: causationID,
+       }
+
+       err := s.publisher.Availability(ctx, s.settings.DeviceID, protocol.Metadata{SentAt: &observedAt}, protocol.AvailabilityPayload{Status: "online"})
+       if err != nil {
+               return fmt.Errorf("publish thermostat contact: %w", err)
+       }
+
+       value, err := json.Marshal(newReportedControl(state))
+       if err != nil {
+               return fmt.Errorf("encode thermostat feedback: %w", err)
+       }
+
+       err = s.publisher.Reported(ctx, s.settings.DeviceID, "control", metadata, protocol.ValuePayload{Value: value})
+       if err != nil {
+               return fmt.Errorf("publish thermostat feedback: %w", err)
+       }
+
+       return nil
+}
+
+func (s *thermostatState) miss(ctx context.Context) error {
+       s.missed = min(s.missed+1, 3)
+       if s.missed < 3 {
+               return nil
+       }
+
+       err := s.publisher.Availability(ctx, s.settings.DeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"})
+       if err != nil {
+               return fmt.Errorf("publish offline: %w", err)
+       }
+
+       return nil
+}
+
+func (s *thermostatState) result(ctx context.Context, command driver.Command, payload protocol.ResultPayload) error {
+       record, err := s.ledger.Remember(command.Metadata.MessageID, payload)
+       if err != nil {
+               return fmt.Errorf("remember thermostat result: %w", err)
+       }
+
+       err = s.publisher.Result(ctx, s.settings.DeviceID, command.Metadata.MessageID, record.Metadata, record.Payload)
+       if err != nil {
+               return fmt.Errorf("publish thermostat result: %w", err)
+       }
+
+       return nil
+}