// 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" "time" devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control" "git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/metrics" ) // 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, "turn_on", 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, "turn_off", 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, "set_level", 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, action, deviceID string, desiredPatch, commandPayload map[string]any) (*devicecontrol.CommandResult, error) { start := time.Now() result := s.doDispatch(ctx, action, deviceID, desiredPatch, commandPayload) metrics.CommandDuration.WithLabelValues(action).Observe(time.Since(start).Seconds()) outcome := "success" if !result.Success { outcome = "error" } metrics.CommandsTotal.WithLabelValues(action, outcome).Inc() return result, nil } func (s *Server) doDispatch(ctx context.Context, action, deviceID string, desiredPatch, commandPayload map[string]any) *devicecontrol.CommandResult { if deviceID == "" { return &devicecontrol.CommandResult{Success: false, Error: "device_id is required"} } if err := s.shadow.PatchDesiredState(ctx, deviceID, desiredPatch); err != nil { s.logger.Error("patch desired state failed", "action", action, "device_id", deviceID, "error", err) return &devicecontrol.CommandResult{Success: false, Error: "failed to update desired state"} } body, err := json.Marshal(commandPayload) if err != nil { s.logger.Error("encode command failed", "action", action, "device_id", deviceID, "error", err) return &devicecontrol.CommandResult{Success: false, Error: "failed to encode command"} } topic := fmt.Sprintf("devices/%s/commands", deviceID) if err := s.mqtt.Publish(topic, 1, false, body); err != nil { s.logger.Error("publish command failed", "action", action, "device_id", deviceID, "topic", topic, "error", err) return &devicecontrol.CommandResult{Success: false, Error: "failed to dispatch command"} } return &devicecontrol.CommandResult{Success: true} }