Реализация device-control-service: gRPC + MQTT + Device Shadow в Redis
Добавлен gRPC-контракт proto/device_control (TurnOn/TurnOff/SetLevel), общий Go-модуль proto/ для переиспользования сгенерированного кода. device-control-service принимает команды по gRPC, обновляет desired_state в Redis и публикует их в MQTT; слушает devices/+/telemetry и devices/+/ack для отметки живости устройства и обновления reported_state; горутина health-check переводит устройство в offline по таймауту и публикует событие в RabbitMQ (device.status_changed) для будущего notification-service. Проверено сквозным тестом через docker compose: grpcurl TurnOn/SetLevel меняет desired_state и уходит в MQTT, mosquitto_pub с ack обновляет reported_state и статус, health-check автоматически переводит устройство в offline по истечении таймаута.
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
// Package config loads device-control-service settings from environment
|
||||
// variables, matching the names used in the repo-root .env.example.
|
||||
package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
GRPCPort int
|
||||
|
||||
MQTTBrokerURL string
|
||||
MQTTClientID string
|
||||
|
||||
RedisAddr string
|
||||
RedisPassword string
|
||||
RedisDB int
|
||||
|
||||
RabbitMQURL string
|
||||
|
||||
HealthCheckTimeout time.Duration
|
||||
HealthCheckInterval time.Duration
|
||||
}
|
||||
|
||||
func Load() (Config, error) {
|
||||
cfg := Config{
|
||||
MQTTBrokerURL: fmt.Sprintf("tcp://%s:%s", getEnv("MQTT_HOST", "localhost"), getEnv("MQTT_PORT", "1883")),
|
||||
MQTTClientID: getEnv("DEVICE_CONTROL_MQTT_CLIENT_ID", "device-control-service"),
|
||||
|
||||
RedisAddr: fmt.Sprintf("%s:%s", getEnv("REDIS_HOST", "localhost"), getEnv("REDIS_PORT", "6379")),
|
||||
RedisPassword: getEnv("REDIS_PASSWORD", ""),
|
||||
|
||||
RabbitMQURL: fmt.Sprintf("amqp://%s:%s@%s:%s/",
|
||||
getEnv("RABBITMQ_USER", "guest"),
|
||||
getEnv("RABBITMQ_PASSWORD", "guest"),
|
||||
getEnv("RABBITMQ_HOST", "localhost"),
|
||||
getEnv("RABBITMQ_PORT", "5672"),
|
||||
),
|
||||
}
|
||||
|
||||
var err error
|
||||
if cfg.GRPCPort, err = getEnvInt("DEVICE_CONTROL_GRPC_PORT", 50051); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.RedisDB, err = getEnvInt("REDIS_DB", 0); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.HealthCheckTimeout, err = getEnvDuration("DEVICE_CONTROL_HEALTHCHECK_TIMEOUT", 60*time.Second); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.HealthCheckInterval, err = getEnvDuration("DEVICE_CONTROL_HEALTHCHECK_INTERVAL", 15*time.Second); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func getEnv(key, fallback string) string {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return v
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
func getEnvInt(key string, fallback int) (int, error) {
|
||||
v := os.Getenv(key)
|
||||
if v == "" {
|
||||
return fallback, nil
|
||||
}
|
||||
n, err := strconv.Atoi(v)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s: %w", key, err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func getEnvDuration(key string, fallback time.Duration) (time.Duration, error) {
|
||||
v := os.Getenv(key)
|
||||
if v == "" {
|
||||
return fallback, nil
|
||||
}
|
||||
d, err := time.ParseDuration(v)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s: %w", key, err)
|
||||
}
|
||||
return d, nil
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
// Package healthcheck periodically flips devices to offline when they stop
|
||||
// sending telemetry/acks, per the platform's Device Shadow health-check
|
||||
// design (a goroutine + graceful shutdown, deliberately simple).
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DeviceStore is the subset of shadow.Store the checker needs, kept as an
|
||||
// interface so tests don't require a real Redis.
|
||||
type DeviceStore interface {
|
||||
KnownDevices(ctx context.Context) ([]string, error)
|
||||
MarkOfflineIfStale(ctx context.Context, deviceID string, timeout time.Duration) (bool, error)
|
||||
}
|
||||
|
||||
// OnOfflineFunc is called once per device that just transitioned to offline.
|
||||
type OnOfflineFunc func(deviceID string)
|
||||
|
||||
type Checker struct {
|
||||
store DeviceStore
|
||||
timeout time.Duration
|
||||
interval time.Duration
|
||||
onOffline OnOfflineFunc
|
||||
logger *slog.Logger
|
||||
|
||||
stop chan struct{}
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func New(store DeviceStore, timeout, interval time.Duration, onOffline OnOfflineFunc, logger *slog.Logger) *Checker {
|
||||
return &Checker{
|
||||
store: store,
|
||||
timeout: timeout,
|
||||
interval: interval,
|
||||
onOffline: onOffline,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the periodic scan loop. Call once.
|
||||
func (c *Checker) Start() {
|
||||
c.stop = make(chan struct{})
|
||||
c.done = make(chan struct{})
|
||||
go c.loop()
|
||||
}
|
||||
|
||||
// Stop signals the loop to exit and waits for it to finish.
|
||||
func (c *Checker) Stop() {
|
||||
close(c.stop)
|
||||
<-c.done
|
||||
}
|
||||
|
||||
func (c *Checker) loop() {
|
||||
defer close(c.done)
|
||||
ticker := time.NewTicker(c.interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
c.checkAll()
|
||||
case <-c.stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Checker) checkAll() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), c.interval)
|
||||
defer cancel()
|
||||
|
||||
ids, err := c.store.KnownDevices(ctx)
|
||||
if err != nil {
|
||||
c.logger.Error("healthcheck: list known devices failed", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
for _, id := range ids {
|
||||
changed, err := c.store.MarkOfflineIfStale(ctx, id, c.timeout)
|
||||
if err != nil {
|
||||
c.logger.Error("healthcheck: check device failed", "device_id", id, "error", err)
|
||||
continue
|
||||
}
|
||||
if changed {
|
||||
c.logger.Info("device marked offline", "device_id", id)
|
||||
if c.onOffline != nil {
|
||||
c.onOffline(id)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type fakeStore struct {
|
||||
mu sync.Mutex
|
||||
known []string
|
||||
stale map[string]bool // deviceID -> whether MarkOfflineIfStale should report a change
|
||||
changes []string // devices actually marked changed, in call order
|
||||
}
|
||||
|
||||
func (f *fakeStore) KnownDevices(context.Context) ([]string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return append([]string(nil), f.known...), nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) MarkOfflineIfStale(_ context.Context, deviceID string, _ time.Duration) (bool, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if !f.stale[deviceID] {
|
||||
return false, nil
|
||||
}
|
||||
// Only report the transition once, like the real store would.
|
||||
f.stale[deviceID] = false
|
||||
f.changes = append(f.changes, deviceID)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func discardLogger() *slog.Logger {
|
||||
return slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
}
|
||||
|
||||
func TestChecker_FlipsStaleDevicesAndCallsOnOffline(t *testing.T) {
|
||||
store := &fakeStore{
|
||||
known: []string{"d1", "d2"},
|
||||
stale: map[string]bool{"d1": true, "d2": false},
|
||||
}
|
||||
|
||||
offline := make(chan string, 2)
|
||||
c := New(store, time.Minute, 10*time.Millisecond, func(deviceID string) {
|
||||
offline <- deviceID
|
||||
}, discardLogger())
|
||||
c.Start()
|
||||
defer c.Stop()
|
||||
|
||||
select {
|
||||
case id := <-offline:
|
||||
if id != "d1" {
|
||||
t.Fatalf("got offline callback for %q, want d1", id)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for onOffline callback")
|
||||
}
|
||||
|
||||
select {
|
||||
case id := <-offline:
|
||||
t.Fatalf("unexpected second onOffline callback for %q (d2 was never stale)", id)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
// expected: no further callbacks
|
||||
}
|
||||
}
|
||||
|
||||
func TestChecker_StopWaitsForLoopExit(t *testing.T) {
|
||||
store := &fakeStore{known: nil}
|
||||
c := New(store, time.Minute, time.Hour, nil, discardLogger())
|
||||
c.Start()
|
||||
c.Stop() // must return without hanging
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
// Package mqttclient wraps paho for device-control-service's two-way MQTT
|
||||
// use: subscribing to telemetry/ack topics for liveness and reported state,
|
||||
// and publishing commands to devices.
|
||||
package mqttclient
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||||
)
|
||||
|
||||
// Handler receives one message's topic, raw payload, and arrival time.
|
||||
type Handler func(topic string, payload []byte, receivedAt time.Time)
|
||||
|
||||
type Config struct {
|
||||
BrokerURL string // e.g. tcp://mosquitto:1883
|
||||
ClientID string
|
||||
}
|
||||
|
||||
type subscription struct {
|
||||
topic string
|
||||
qos byte
|
||||
handler Handler
|
||||
}
|
||||
|
||||
// Client wraps a paho client, replaying every registered subscription on
|
||||
// each (re)connect so a dropped connection resubscribes automatically.
|
||||
type Client struct {
|
||||
client mqtt.Client
|
||||
logger *slog.Logger
|
||||
|
||||
mu sync.Mutex
|
||||
subs []subscription
|
||||
}
|
||||
|
||||
func Connect(cfg Config, logger *slog.Logger) (*Client, error) {
|
||||
c := &Client{logger: logger}
|
||||
|
||||
opts := mqtt.NewClientOptions().
|
||||
AddBroker(cfg.BrokerURL).
|
||||
SetClientID(cfg.ClientID).
|
||||
SetAutoReconnect(true).
|
||||
SetConnectRetry(true).
|
||||
SetOnConnectHandler(func(mc mqtt.Client) {
|
||||
c.resubscribeAll(mc)
|
||||
}).
|
||||
SetConnectionLostHandler(func(_ mqtt.Client, err error) {
|
||||
logger.Warn("mqtt connection lost", "error", err)
|
||||
})
|
||||
|
||||
mqttClient := mqtt.NewClient(opts)
|
||||
token := mqttClient.Connect()
|
||||
token.Wait()
|
||||
if err := token.Error(); err != nil {
|
||||
return nil, fmt.Errorf("connect to mqtt broker %s: %w", cfg.BrokerURL, err)
|
||||
}
|
||||
c.client = mqttClient
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// Subscribe registers a handler for topic. It subscribes immediately if
|
||||
// already connected, and will be replayed on every future reconnect.
|
||||
func (c *Client) Subscribe(topic string, qos byte, handler Handler) error {
|
||||
c.mu.Lock()
|
||||
c.subs = append(c.subs, subscription{topic: topic, qos: qos, handler: handler})
|
||||
c.mu.Unlock()
|
||||
|
||||
if !c.client.IsConnectionOpen() {
|
||||
return nil
|
||||
}
|
||||
return c.subscribeOne(c.client, subscription{topic: topic, qos: qos, handler: handler})
|
||||
}
|
||||
|
||||
func (c *Client) resubscribeAll(mc mqtt.Client) {
|
||||
c.mu.Lock()
|
||||
subs := append([]subscription(nil), c.subs...)
|
||||
c.mu.Unlock()
|
||||
|
||||
for _, s := range subs {
|
||||
if err := c.subscribeOne(mc, s); err != nil {
|
||||
c.logger.Error("mqtt subscribe failed", "topic", s.topic, "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) subscribeOne(mc mqtt.Client, s subscription) error {
|
||||
token := mc.Subscribe(s.topic, s.qos, func(_ mqtt.Client, msg mqtt.Message) {
|
||||
s.handler(msg.Topic(), msg.Payload(), time.Now())
|
||||
})
|
||||
token.Wait()
|
||||
return token.Error()
|
||||
}
|
||||
|
||||
// Publish sends payload to topic and waits for the publish to complete.
|
||||
func (c *Client) Publish(topic string, qos byte, retained bool, payload []byte) error {
|
||||
token := c.client.Publish(topic, qos, retained, payload)
|
||||
token.Wait()
|
||||
return token.Error()
|
||||
}
|
||||
|
||||
// Close disconnects, waiting up to 250ms for in-flight work to settle.
|
||||
func (c *Client) Close() {
|
||||
c.client.Disconnect(250)
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
// Package rabbitmq publishes device online/offline transitions for
|
||||
// notification-service to consume. Delivery reliability matters more than
|
||||
// latency here, so messages are persistent and the queue is durable.
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
)
|
||||
|
||||
const StatusChangedQueue = "device.status_changed"
|
||||
|
||||
type Publisher struct {
|
||||
conn *amqp.Connection
|
||||
ch *amqp.Channel
|
||||
}
|
||||
|
||||
func Connect(url string) (*Publisher, error) {
|
||||
conn, err := amqp.Dial(url)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("dial rabbitmq: %w", err)
|
||||
}
|
||||
|
||||
ch, err := conn.Channel()
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("open channel: %w", err)
|
||||
}
|
||||
|
||||
if _, err := ch.QueueDeclare(StatusChangedQueue, true, false, false, false, nil); err != nil {
|
||||
ch.Close()
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("declare queue %s: %w", StatusChangedQueue, err)
|
||||
}
|
||||
|
||||
return &Publisher{conn: conn, ch: ch}, nil
|
||||
}
|
||||
|
||||
func (p *Publisher) Close() error {
|
||||
if err := p.ch.Close(); err != nil {
|
||||
p.conn.Close()
|
||||
return err
|
||||
}
|
||||
return p.conn.Close()
|
||||
}
|
||||
|
||||
type statusChangedEvent struct {
|
||||
DeviceID string `json:"device_id"`
|
||||
Status string `json:"status"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// PublishStatusChanged emits one event when a device transitions online or offline.
|
||||
func (p *Publisher) PublishStatusChanged(ctx context.Context, deviceID, status string) error {
|
||||
body, err := json.Marshal(statusChangedEvent{
|
||||
DeviceID: deviceID,
|
||||
Status: status,
|
||||
At: time.Now(),
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal event: %w", err)
|
||||
}
|
||||
|
||||
err = p.ch.PublishWithContext(ctx, "", StatusChangedQueue, false, false, amqp.Publishing{
|
||||
ContentType: "application/json",
|
||||
DeliveryMode: amqp.Persistent,
|
||||
Timestamp: time.Now(),
|
||||
Body: body,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("publish status change for device %s: %w", deviceID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
// Package server implements the DeviceControl gRPC service: it patches the
|
||||
// device's desired state in Redis and dispatches the command over MQTT.
|
||||
// It deliberately does not wait for the device to execute the command —
|
||||
// see the Device Shadow note in the service README.
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
|
||||
)
|
||||
|
||||
// ShadowPatcher is the subset of shadow.Store the server needs.
|
||||
type ShadowPatcher interface {
|
||||
PatchDesiredState(ctx context.Context, deviceID string, patch map[string]any) error
|
||||
}
|
||||
|
||||
// CommandPublisher is the subset of mqttclient.Client the server needs.
|
||||
type CommandPublisher interface {
|
||||
Publish(topic string, qos byte, retained bool, payload []byte) error
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
devicecontrol.UnimplementedDeviceControlServer
|
||||
|
||||
shadow ShadowPatcher
|
||||
mqtt CommandPublisher
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
func New(shadow ShadowPatcher, mqtt CommandPublisher, logger *slog.Logger) *Server {
|
||||
return &Server{shadow: shadow, mqtt: mqtt, logger: logger}
|
||||
}
|
||||
|
||||
func (s *Server) TurnOn(ctx context.Context, req *devicecontrol.TurnOnRequest) (*devicecontrol.CommandResult, error) {
|
||||
return s.dispatch(ctx, req.GetDeviceId(),
|
||||
map[string]any{"power": "on"},
|
||||
map[string]any{"action": "turn_on"},
|
||||
)
|
||||
}
|
||||
|
||||
func (s *Server) TurnOff(ctx context.Context, req *devicecontrol.TurnOffRequest) (*devicecontrol.CommandResult, error) {
|
||||
return s.dispatch(ctx, req.GetDeviceId(),
|
||||
map[string]any{"power": "off"},
|
||||
map[string]any{"action": "turn_off"},
|
||||
)
|
||||
}
|
||||
|
||||
func (s *Server) SetLevel(ctx context.Context, req *devicecontrol.SetLevelRequest) (*devicecontrol.CommandResult, error) {
|
||||
return s.dispatch(ctx, req.GetDeviceId(),
|
||||
map[string]any{"level": req.GetLevel()},
|
||||
map[string]any{"action": "set_level", "level": req.GetLevel()},
|
||||
)
|
||||
}
|
||||
|
||||
// dispatch is shared by all three RPCs: patch desired state first (so the UI
|
||||
// gets an instant, optimistic view even if the device is offline), then hand
|
||||
// the command to MQTT. Business failures come back as CommandResult.Error,
|
||||
// not a gRPC error — the wire contract is success/error in one message.
|
||||
func (s *Server) dispatch(ctx context.Context, deviceID string, desiredPatch, commandPayload map[string]any) (*devicecontrol.CommandResult, error) {
|
||||
if deviceID == "" {
|
||||
return &devicecontrol.CommandResult{Success: false, Error: "device_id is required"}, nil
|
||||
}
|
||||
|
||||
if err := s.shadow.PatchDesiredState(ctx, deviceID, desiredPatch); err != nil {
|
||||
s.logger.Error("patch desired state failed", "device_id", deviceID, "error", err)
|
||||
return &devicecontrol.CommandResult{Success: false, Error: "failed to update desired state"}, nil
|
||||
}
|
||||
|
||||
body, err := json.Marshal(commandPayload)
|
||||
if err != nil {
|
||||
s.logger.Error("encode command failed", "device_id", deviceID, "error", err)
|
||||
return &devicecontrol.CommandResult{Success: false, Error: "failed to encode command"}, nil
|
||||
}
|
||||
|
||||
topic := fmt.Sprintf("devices/%s/commands", deviceID)
|
||||
if err := s.mqtt.Publish(topic, 1, false, body); err != nil {
|
||||
s.logger.Error("publish command failed", "device_id", deviceID, "topic", topic, "error", err)
|
||||
return &devicecontrol.CommandResult{Success: false, Error: "failed to dispatch command"}, nil
|
||||
}
|
||||
|
||||
return &devicecontrol.CommandResult{Success: true}, nil
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"testing"
|
||||
|
||||
devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
|
||||
)
|
||||
|
||||
type fakeShadow struct {
|
||||
err error
|
||||
patches []map[string]any
|
||||
}
|
||||
|
||||
func (f *fakeShadow) PatchDesiredState(_ context.Context, _ string, patch map[string]any) error {
|
||||
if f.err != nil {
|
||||
return f.err
|
||||
}
|
||||
f.patches = append(f.patches, patch)
|
||||
return nil
|
||||
}
|
||||
|
||||
type fakeMQTT struct {
|
||||
err error
|
||||
topic string
|
||||
payload []byte
|
||||
}
|
||||
|
||||
func (f *fakeMQTT) Publish(topic string, _ byte, _ bool, payload []byte) error {
|
||||
if f.err != nil {
|
||||
return f.err
|
||||
}
|
||||
f.topic = topic
|
||||
f.payload = payload
|
||||
return nil
|
||||
}
|
||||
|
||||
func discardLogger() *slog.Logger {
|
||||
return slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
}
|
||||
|
||||
func TestTurnOn_Success(t *testing.T) {
|
||||
sh := &fakeShadow{}
|
||||
mq := &fakeMQTT{}
|
||||
s := New(sh, mq, discardLogger())
|
||||
|
||||
res, err := s.TurnOn(context.Background(), &devicecontrol.TurnOnRequest{DeviceId: "d1"})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !res.Success {
|
||||
t.Fatalf("expected success, got error %q", res.Error)
|
||||
}
|
||||
if len(sh.patches) != 1 || sh.patches[0]["power"] != "on" {
|
||||
t.Fatalf("got desired patches %+v, want [{power: on}]", sh.patches)
|
||||
}
|
||||
if mq.topic != "devices/d1/commands" {
|
||||
t.Fatalf("got topic %q, want devices/d1/commands", mq.topic)
|
||||
}
|
||||
var payload map[string]any
|
||||
if err := json.Unmarshal(mq.payload, &payload); err != nil {
|
||||
t.Fatalf("unmarshal published payload: %v", err)
|
||||
}
|
||||
if payload["action"] != "turn_on" {
|
||||
t.Fatalf("got payload %+v, want action=turn_on", payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetLevel_Success(t *testing.T) {
|
||||
sh := &fakeShadow{}
|
||||
mq := &fakeMQTT{}
|
||||
s := New(sh, mq, discardLogger())
|
||||
|
||||
res, err := s.SetLevel(context.Background(), &devicecontrol.SetLevelRequest{DeviceId: "d1", Level: 42.5})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !res.Success {
|
||||
t.Fatalf("expected success, got error %q", res.Error)
|
||||
}
|
||||
if sh.patches[0]["level"] != 42.5 {
|
||||
t.Fatalf("got desired patch %+v, want level=42.5", sh.patches[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_MissingDeviceID(t *testing.T) {
|
||||
s := New(&fakeShadow{}, &fakeMQTT{}, discardLogger())
|
||||
|
||||
res, err := s.TurnOff(context.Background(), &devicecontrol.TurnOffRequest{DeviceId: ""})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected transport error: %v", err)
|
||||
}
|
||||
if res.Success {
|
||||
t.Fatal("expected failure for missing device_id")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_ShadowFailure(t *testing.T) {
|
||||
s := New(&fakeShadow{err: errors.New("redis down")}, &fakeMQTT{}, discardLogger())
|
||||
|
||||
res, err := s.TurnOn(context.Background(), &devicecontrol.TurnOnRequest{DeviceId: "d1"})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected transport error: %v", err)
|
||||
}
|
||||
if res.Success {
|
||||
t.Fatal("expected failure when shadow patch errors")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatch_PublishFailure(t *testing.T) {
|
||||
s := New(&fakeShadow{}, &fakeMQTT{err: errors.New("broker unreachable")}, discardLogger())
|
||||
|
||||
res, err := s.TurnOn(context.Background(), &devicecontrol.TurnOnRequest{DeviceId: "d1"})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected transport error: %v", err)
|
||||
}
|
||||
if res.Success {
|
||||
t.Fatal("expected failure when mqtt publish errors")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
// Package shadow implements the Device Shadow pattern on top of Redis:
|
||||
// desired_state (what the user wants) vs reported_state (what the device
|
||||
// confirmed), plus status and last_seen for liveness tracking.
|
||||
package shadow
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// knownDevicesKey backs a lightweight device registry: health-check needs to
|
||||
// know which device IDs to scan, and this service has no PostgreSQL access
|
||||
// (device registration lives in Laravel), so every device we hear from gets
|
||||
// added to this set.
|
||||
const knownDevicesKey = "devices:known"
|
||||
|
||||
func desiredKey(deviceID string) string { return "device:" + deviceID + ":desired_state" }
|
||||
func reportedKey(deviceID string) string { return "device:" + deviceID + ":reported_state" }
|
||||
func statusKey(deviceID string) string { return "device:" + deviceID + ":status" }
|
||||
func lastSeenKey(deviceID string) string { return "device:" + deviceID + ":last_seen" }
|
||||
|
||||
const (
|
||||
StatusOnline = "online"
|
||||
StatusOffline = "offline"
|
||||
)
|
||||
|
||||
type Store struct {
|
||||
rdb *redis.Client
|
||||
}
|
||||
|
||||
func New(rdb *redis.Client) *Store {
|
||||
return &Store{rdb: rdb}
|
||||
}
|
||||
|
||||
func Connect(ctx context.Context, addr, password string, db int) (*Store, error) {
|
||||
rdb := redis.NewClient(&redis.Options{Addr: addr, Password: password, DB: db})
|
||||
if err := rdb.Ping(ctx).Err(); err != nil {
|
||||
return nil, fmt.Errorf("ping redis: %w", err)
|
||||
}
|
||||
return New(rdb), nil
|
||||
}
|
||||
|
||||
func (s *Store) Close() error {
|
||||
return s.rdb.Close()
|
||||
}
|
||||
|
||||
// PatchDesiredState merges patch into the device's desired_state JSON object.
|
||||
func (s *Store) PatchDesiredState(ctx context.Context, deviceID string, patch map[string]any) error {
|
||||
return s.patchJSON(ctx, desiredKey(deviceID), patch)
|
||||
}
|
||||
|
||||
// PatchReportedState merges patch into the device's reported_state JSON object.
|
||||
func (s *Store) PatchReportedState(ctx context.Context, deviceID string, patch map[string]any) error {
|
||||
return s.patchJSON(ctx, reportedKey(deviceID), patch)
|
||||
}
|
||||
|
||||
func (s *Store) patchJSON(ctx context.Context, key string, patch map[string]any) error {
|
||||
existing := map[string]any{}
|
||||
raw, err := s.rdb.Get(ctx, key).Result()
|
||||
switch {
|
||||
case err == redis.Nil:
|
||||
// no existing state yet, start from an empty object
|
||||
case err != nil:
|
||||
return fmt.Errorf("get %s: %w", key, err)
|
||||
default:
|
||||
if err := json.Unmarshal([]byte(raw), &existing); err != nil {
|
||||
return fmt.Errorf("unmarshal existing %s: %w", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
for k, v := range patch {
|
||||
existing[k] = v
|
||||
}
|
||||
|
||||
merged, err := json.Marshal(existing)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal %s: %w", key, err)
|
||||
}
|
||||
if err := s.rdb.Set(ctx, key, merged, 0).Err(); err != nil {
|
||||
return fmt.Errorf("set %s: %w", key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Touch registers deviceID as known, bumps last_seen to now, and — if the
|
||||
// device wasn't already marked online — flips status to online. It reports
|
||||
// whether that online transition happened, so the caller can decide whether
|
||||
// a "device back online" event is worth emitting.
|
||||
func (s *Store) Touch(ctx context.Context, deviceID string) (becameOnline bool, err error) {
|
||||
if err := s.rdb.SAdd(ctx, knownDevicesKey, deviceID).Err(); err != nil {
|
||||
return false, fmt.Errorf("register known device %s: %w", deviceID, err)
|
||||
}
|
||||
if err := s.rdb.Set(ctx, lastSeenKey(deviceID), time.Now().Unix(), 0).Err(); err != nil {
|
||||
return false, fmt.Errorf("set last_seen for %s: %w", deviceID, err)
|
||||
}
|
||||
|
||||
status, err := s.rdb.Get(ctx, statusKey(deviceID)).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return false, fmt.Errorf("get status for %s: %w", deviceID, err)
|
||||
}
|
||||
if status == StatusOnline {
|
||||
return false, nil
|
||||
}
|
||||
if err := s.rdb.Set(ctx, statusKey(deviceID), StatusOnline, 0).Err(); err != nil {
|
||||
return false, fmt.Errorf("set status online for %s: %w", deviceID, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// MarkOfflineIfStale flips deviceID to offline if it is currently online and
|
||||
// its last_seen is older than timeout. It reports whether a change was made.
|
||||
func (s *Store) MarkOfflineIfStale(ctx context.Context, deviceID string, timeout time.Duration) (changed bool, err error) {
|
||||
status, err := s.rdb.Get(ctx, statusKey(deviceID)).Result()
|
||||
if err == redis.Nil || status != StatusOnline {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("get status for %s: %w", deviceID, err)
|
||||
}
|
||||
|
||||
lastSeenRaw, err := s.rdb.Get(ctx, lastSeenKey(deviceID)).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return false, fmt.Errorf("get last_seen for %s: %w", deviceID, err)
|
||||
}
|
||||
|
||||
var lastSeen time.Time
|
||||
if lastSeenRaw != "" {
|
||||
sec, err := strconv.ParseInt(lastSeenRaw, 10, 64)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("parse last_seen for %s: %w", deviceID, err)
|
||||
}
|
||||
lastSeen = time.Unix(sec, 0)
|
||||
}
|
||||
|
||||
if time.Since(lastSeen) <= timeout {
|
||||
return false, nil
|
||||
}
|
||||
if err := s.rdb.Set(ctx, statusKey(deviceID), StatusOffline, 0).Err(); err != nil {
|
||||
return false, fmt.Errorf("set status offline for %s: %w", deviceID, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// KnownDevices lists every device ID this instance has ever heard from.
|
||||
func (s *Store) KnownDevices(ctx context.Context) ([]string, error) {
|
||||
ids, err := s.rdb.SMembers(ctx, knownDevicesKey).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list known devices: %w", err)
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
package shadow
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func newTestStore(t *testing.T) *Store {
|
||||
t.Helper()
|
||||
mr := miniredis.RunT(t)
|
||||
rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
||||
t.Cleanup(func() { rdb.Close() })
|
||||
return New(rdb)
|
||||
}
|
||||
|
||||
func TestPatchDesiredState_MergesFields(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newTestStore(t)
|
||||
|
||||
if err := s.PatchDesiredState(ctx, "d1", map[string]any{"power": "on"}); err != nil {
|
||||
t.Fatalf("first patch: %v", err)
|
||||
}
|
||||
if err := s.PatchDesiredState(ctx, "d1", map[string]any{"level": 42.0}); err != nil {
|
||||
t.Fatalf("second patch: %v", err)
|
||||
}
|
||||
|
||||
raw, err := s.rdb.Get(ctx, desiredKey("d1")).Result()
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
var got map[string]any
|
||||
if err := json.Unmarshal([]byte(raw), &got); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if got["power"] != "on" || got["level"] != 42.0 {
|
||||
t.Fatalf("got %+v, want power=on and level=42", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTouch_FirstCallReportsOnlineTransition(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newTestStore(t)
|
||||
|
||||
becameOnline, err := s.Touch(ctx, "d1")
|
||||
if err != nil {
|
||||
t.Fatalf("touch: %v", err)
|
||||
}
|
||||
if !becameOnline {
|
||||
t.Fatal("expected first touch to report an online transition")
|
||||
}
|
||||
|
||||
becameOnline, err = s.Touch(ctx, "d1")
|
||||
if err != nil {
|
||||
t.Fatalf("second touch: %v", err)
|
||||
}
|
||||
if becameOnline {
|
||||
t.Fatal("expected second touch (already online) to report no transition")
|
||||
}
|
||||
|
||||
ids, err := s.KnownDevices(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("known devices: %v", err)
|
||||
}
|
||||
if len(ids) != 1 || ids[0] != "d1" {
|
||||
t.Fatalf("got known devices %+v, want [d1]", ids)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkOfflineIfStale(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newTestStore(t)
|
||||
|
||||
if _, err := s.Touch(ctx, "d1"); err != nil {
|
||||
t.Fatalf("touch: %v", err)
|
||||
}
|
||||
|
||||
changed, err := s.MarkOfflineIfStale(ctx, "d1", time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("mark offline (not stale): %v", err)
|
||||
}
|
||||
if changed {
|
||||
t.Fatal("device just touched should not be considered stale")
|
||||
}
|
||||
|
||||
changed, err = s.MarkOfflineIfStale(ctx, "d1", -time.Second) // any age counts as stale
|
||||
if err != nil {
|
||||
t.Fatalf("mark offline (stale): %v", err)
|
||||
}
|
||||
if !changed {
|
||||
t.Fatal("expected status to flip to offline")
|
||||
}
|
||||
|
||||
// Already offline: a second call should report no further change.
|
||||
changed, err = s.MarkOfflineIfStale(ctx, "d1", -time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("mark offline (already offline): %v", err)
|
||||
}
|
||||
if changed {
|
||||
t.Fatal("expected no change once already offline")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkOfflineIfStale_UnknownDeviceIsNoop(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newTestStore(t)
|
||||
|
||||
changed, err := s.MarkOfflineIfStale(ctx, "ghost", time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if changed {
|
||||
t.Fatal("unknown device should never be reported as changed")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user