From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 23:53:56 +0000 (-0300) Subject: Poll and command shared Alamo thermostats X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/refs/heads/master?ds=inline;hp=b9a4f3bdd2609a8b178dd42f313c383ccb59049b 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 69b14d2..55f1dab 100644 --- 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 -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 diff --git a/doc/alamo.rst b/doc/alamo.rst index c9e6fde..8c02dce 100644 --- 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. + +**************** + 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 de70846..29cfb41 100644 --- 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 -// 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) diff --git a/driver/alamo/command_handler.go b/driver/alamo/command_handler.go new file mode 100644 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 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 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 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 03d36f2..bef18e2 100644 --- a/driver/alamo/fake_helper_test.go +++ b/driver/alamo/fake_helper_test.go @@ -2,20 +2,43 @@ package alamo import ( "context" + "errors" + "fmt" + "sync" + "testing" "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 { - 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) { + g.mu.Lock() + defer g.mu.Unlock() + 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 } +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) { + g.mu.Lock() + defer g.mu.Unlock() + + g.actions = append(g.actions, action) + 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] @@ -38,11 +84,11 @@ func (g *fakeGateway) Action(ctx context.Context, mac string, action Action) (St continue } - if action.Power != nil { + if action.Power != nil && !g.ignorePower { state.Power = *action.Power } - if action.TargetTemperature != nil { + if action.TargetTemperature != nil && !g.ignoreTarget { state.TargetTemperature = *action.TargetTemperature } @@ -52,20 +98,110 @@ func (g *fakeGateway) Action(ctx context.Context, mac string, action Action) (St 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 { diff --git a/driver/alamo/gateway.go b/driver/alamo/gateway.go index 1bf0074..2f3da3b 100644 --- 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 } -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 { @@ -73,7 +56,29 @@ func (g *gateway) wait(ctx context.Context) error { return err } + deadline, exists := ctx.Deadline() + if exists && !time.Now().Before(deadline) { + return context.DeadlineExceeded + } + 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 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 715e860..ab004e0 100644 --- 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) }) } + +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 7179124..4c645f4 100644 --- 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) } -// 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() @@ -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 { - 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() { @@ -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 { - 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 { @@ -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 { - 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] { @@ -167,3 +120,70 @@ func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values 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 7ee0e36..1c8f8a0 100644 --- a/driver/alamo/http_client_test.go +++ b/driver/alamo/http_client_test.go @@ -15,8 +15,6 @@ import ( "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 @@ -251,6 +249,18 @@ func TestHTTPClientDoesNotFollowRedirects(t *testing.T) { 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) { diff --git a/driver/alamo/queued_commands.go b/driver/alamo/queued_commands.go new file mode 100644 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 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 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 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 +}