home_automatization/services/device-control-service/internal/server/server.go

// 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}
}