From 3eb84aae6c3076b74b8fd8394c81a83b6dfd6939 Mon Sep 17 00:00:00 2001 From: Dmitry Gammel Date: Sun, 26 Jul 2026 18:40:36 +0500 Subject: [PATCH] =?UTF-8?q?=D0=A0=D0=B5=D0=B0=D0=BB=D0=B8=D0=B7=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D1=8F=20device-control-service:=20gRPC=20+=20MQTT?= =?UTF-8?q?=20+=20Device=20Shadow=20=D0=B2=20Redis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Добавлен 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 по истечении таймаута. --- .env.example | 4 + .gitignore | 5 + docker-compose.yml | 37 ++- proto/.gitkeep | 0 proto/device_control/device_control.pb.go | 291 ++++++++++++++++++ proto/device_control/device_control.proto | 36 +++ .../device_control/device_control_grpc.pb.go | 213 +++++++++++++ proto/go.mod | 15 + proto/go.sum | 38 +++ services/device-control-service/Dockerfile | 15 + services/device-control-service/README.md | 43 ++- .../cmd/device-control-service/main.go | 171 ++++++++++ services/device-control-service/go.mod | 27 ++ services/device-control-service/go.sum | 70 +++++ .../internal/config/config.go | 90 ++++++ .../internal/healthcheck/checker.go | 93 ++++++ .../internal/healthcheck/checker_test.go | 76 +++++ .../internal/mqttclient/client.go | 107 +++++++ .../internal/rabbitmq/publisher.go | 78 +++++ .../internal/server/server.go | 86 ++++++ .../internal/server/server_test.go | 124 ++++++++ .../internal/shadow/store.go | 156 ++++++++++ .../internal/shadow/store_test.go | 119 +++++++ 23 files changed, 1885 insertions(+), 9 deletions(-) delete mode 100644 proto/.gitkeep create mode 100644 proto/device_control/device_control.pb.go create mode 100644 proto/device_control/device_control.proto create mode 100644 proto/device_control/device_control_grpc.pb.go create mode 100644 proto/go.mod create mode 100644 proto/go.sum create mode 100644 services/device-control-service/Dockerfile create mode 100644 services/device-control-service/cmd/device-control-service/main.go create mode 100644 services/device-control-service/go.mod create mode 100644 services/device-control-service/go.sum create mode 100644 services/device-control-service/internal/config/config.go create mode 100644 services/device-control-service/internal/healthcheck/checker.go create mode 100644 services/device-control-service/internal/healthcheck/checker_test.go create mode 100644 services/device-control-service/internal/mqttclient/client.go create mode 100644 services/device-control-service/internal/rabbitmq/publisher.go create mode 100644 services/device-control-service/internal/server/server.go create mode 100644 services/device-control-service/internal/server/server_test.go create mode 100644 services/device-control-service/internal/shadow/store.go create mode 100644 services/device-control-service/internal/shadow/store_test.go diff --git a/.env.example b/.env.example index 66628e9..b32d905 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,7 @@ CLICKHOUSE_PASSWORD=change_me REDIS_HOST=redis REDIS_PORT=6379 REDIS_PASSWORD= +REDIS_DB=0 # --- RabbitMQ (async events between Go services) --- RABBITMQ_HOST=rabbitmq @@ -31,6 +32,9 @@ MQTT_PORT=1883 # --- device-control-service (gRPC) --- DEVICE_CONTROL_GRPC_PORT=50051 +DEVICE_CONTROL_MQTT_CLIENT_ID=device-control-service +DEVICE_CONTROL_HEALTHCHECK_TIMEOUT=60s +DEVICE_CONTROL_HEALTHCHECK_INTERVAL=15s # --- ingest-service --- INGEST_MQTT_CLIENT_ID=ingest-service diff --git a/.gitignore b/.gitignore index e3fea65..0cb3ea9 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,11 @@ services/**/bin/ services/**/*.exe *.test *.out +services/ingest-service/ingest-service +services/device-control-service/device-control-service +services/rule-engine-service/rule-engine-service +services/health-check-service/health-check-service +services/esp32-emulator/esp32-emulator # Laravel laravel-app/vendor/ diff --git a/docker-compose.yml b/docker-compose.yml index bb4c826..02a1a03 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -116,6 +116,37 @@ services: - home-automation restart: unless-stopped - # Remaining app services (device-control-service, rule-engine-service, - # health-check-service, esp32-emulator, laravel-app) are added here as they - # get implemented — see services/*/README.md for the plan for each one. + device-control-service: + build: + context: . + dockerfile: services/device-control-service/Dockerfile + ports: + - "${DEVICE_CONTROL_GRPC_PORT}:50051" + environment: + DEVICE_CONTROL_GRPC_PORT: 50051 + MQTT_HOST: mosquitto + MQTT_PORT: 1883 + REDIS_HOST: redis + REDIS_PORT: 6379 + REDIS_PASSWORD: ${REDIS_PASSWORD} + REDIS_DB: ${REDIS_DB} + RABBITMQ_HOST: rabbitmq + RABBITMQ_PORT: 5672 + RABBITMQ_USER: ${RABBITMQ_USER} + RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD} + DEVICE_CONTROL_HEALTHCHECK_TIMEOUT: ${DEVICE_CONTROL_HEALTHCHECK_TIMEOUT} + DEVICE_CONTROL_HEALTHCHECK_INTERVAL: ${DEVICE_CONTROL_HEALTHCHECK_INTERVAL} + depends_on: + mosquitto: + condition: service_started + redis: + condition: service_started + rabbitmq: + condition: service_healthy + networks: + - home-automation + restart: unless-stopped + + # Remaining app services (rule-engine-service, health-check-service, + # esp32-emulator, laravel-app) are added here as they get implemented — + # see services/*/README.md for the plan for each one. diff --git a/proto/.gitkeep b/proto/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/proto/device_control/device_control.pb.go b/proto/device_control/device_control.pb.go new file mode 100644 index 0000000..d692512 --- /dev/null +++ b/proto/device_control/device_control.pb.go @@ -0,0 +1,291 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v7.35.1 +// source: device_control/device_control.proto + +package device_control + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type TurnOnRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + DeviceId string `protobuf:"bytes,1,opt,name=device_id,json=deviceId,proto3" json:"device_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TurnOnRequest) Reset() { + *x = TurnOnRequest{} + mi := &file_device_control_device_control_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TurnOnRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TurnOnRequest) ProtoMessage() {} + +func (x *TurnOnRequest) ProtoReflect() protoreflect.Message { + mi := &file_device_control_device_control_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TurnOnRequest.ProtoReflect.Descriptor instead. +func (*TurnOnRequest) Descriptor() ([]byte, []int) { + return file_device_control_device_control_proto_rawDescGZIP(), []int{0} +} + +func (x *TurnOnRequest) GetDeviceId() string { + if x != nil { + return x.DeviceId + } + return "" +} + +type TurnOffRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + DeviceId string `protobuf:"bytes,1,opt,name=device_id,json=deviceId,proto3" json:"device_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TurnOffRequest) Reset() { + *x = TurnOffRequest{} + mi := &file_device_control_device_control_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TurnOffRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TurnOffRequest) ProtoMessage() {} + +func (x *TurnOffRequest) ProtoReflect() protoreflect.Message { + mi := &file_device_control_device_control_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TurnOffRequest.ProtoReflect.Descriptor instead. +func (*TurnOffRequest) Descriptor() ([]byte, []int) { + return file_device_control_device_control_proto_rawDescGZIP(), []int{1} +} + +func (x *TurnOffRequest) GetDeviceId() string { + if x != nil { + return x.DeviceId + } + return "" +} + +type SetLevelRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + DeviceId string `protobuf:"bytes,1,opt,name=device_id,json=deviceId,proto3" json:"device_id,omitempty"` + Level float64 `protobuf:"fixed64,2,opt,name=level,proto3" json:"level,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SetLevelRequest) Reset() { + *x = SetLevelRequest{} + mi := &file_device_control_device_control_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SetLevelRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SetLevelRequest) ProtoMessage() {} + +func (x *SetLevelRequest) ProtoReflect() protoreflect.Message { + mi := &file_device_control_device_control_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SetLevelRequest.ProtoReflect.Descriptor instead. +func (*SetLevelRequest) Descriptor() ([]byte, []int) { + return file_device_control_device_control_proto_rawDescGZIP(), []int{2} +} + +func (x *SetLevelRequest) GetDeviceId() string { + if x != nil { + return x.DeviceId + } + return "" +} + +func (x *SetLevelRequest) GetLevel() float64 { + if x != nil { + return x.Level + } + return 0 +} + +type CommandResult struct { + state protoimpl.MessageState `protogen:"open.v1"` + Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"` + Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CommandResult) Reset() { + *x = CommandResult{} + mi := &file_device_control_device_control_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CommandResult) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CommandResult) ProtoMessage() {} + +func (x *CommandResult) ProtoReflect() protoreflect.Message { + mi := &file_device_control_device_control_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CommandResult.ProtoReflect.Descriptor instead. +func (*CommandResult) Descriptor() ([]byte, []int) { + return file_device_control_device_control_proto_rawDescGZIP(), []int{3} +} + +func (x *CommandResult) GetSuccess() bool { + if x != nil { + return x.Success + } + return false +} + +func (x *CommandResult) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +var File_device_control_device_control_proto protoreflect.FileDescriptor + +const file_device_control_device_control_proto_rawDesc = "" + + "\n" + + "#device_control/device_control.proto\x12\x0edevice_control\",\n" + + "\rTurnOnRequest\x12\x1b\n" + + "\tdevice_id\x18\x01 \x01(\tR\bdeviceId\"-\n" + + "\x0eTurnOffRequest\x12\x1b\n" + + "\tdevice_id\x18\x01 \x01(\tR\bdeviceId\"D\n" + + "\x0fSetLevelRequest\x12\x1b\n" + + "\tdevice_id\x18\x01 \x01(\tR\bdeviceId\x12\x14\n" + + "\x05level\x18\x02 \x01(\x01R\x05level\"?\n" + + "\rCommandResult\x12\x18\n" + + "\asuccess\x18\x01 \x01(\bR\asuccess\x12\x14\n" + + "\x05error\x18\x02 \x01(\tR\x05error2\xed\x01\n" + + "\rDeviceControl\x12F\n" + + "\x06TurnOn\x12\x1d.device_control.TurnOnRequest\x1a\x1d.device_control.CommandResult\x12H\n" + + "\aTurnOff\x12\x1e.device_control.TurnOffRequest\x1a\x1d.device_control.CommandResult\x12J\n" + + "\bSetLevel\x12\x1f.device_control.SetLevelRequest\x1a\x1d.device_control.CommandResultB>Z device_control.TurnOnRequest + 1, // 1: device_control.DeviceControl.TurnOff:input_type -> device_control.TurnOffRequest + 2, // 2: device_control.DeviceControl.SetLevel:input_type -> device_control.SetLevelRequest + 3, // 3: device_control.DeviceControl.TurnOn:output_type -> device_control.CommandResult + 3, // 4: device_control.DeviceControl.TurnOff:output_type -> device_control.CommandResult + 3, // 5: device_control.DeviceControl.SetLevel:output_type -> device_control.CommandResult + 3, // [3:6] is the sub-list for method output_type + 0, // [0:3] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_device_control_device_control_proto_init() } +func file_device_control_device_control_proto_init() { + if File_device_control_device_control_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_device_control_device_control_proto_rawDesc), len(file_device_control_device_control_proto_rawDesc)), + NumEnums: 0, + NumMessages: 4, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_device_control_device_control_proto_goTypes, + DependencyIndexes: file_device_control_device_control_proto_depIdxs, + MessageInfos: file_device_control_device_control_proto_msgTypes, + }.Build() + File_device_control_device_control_proto = out.File + file_device_control_device_control_proto_goTypes = nil + file_device_control_device_control_proto_depIdxs = nil +} diff --git a/proto/device_control/device_control.proto b/proto/device_control/device_control.proto new file mode 100644 index 0000000..10d6405 --- /dev/null +++ b/proto/device_control/device_control.proto @@ -0,0 +1,36 @@ +syntax = "proto3"; + +package device_control; + +option go_package = "git.cactoz.su/cacto/home_automatization/proto/device_control"; + +// DeviceControl is the synchronous entry point for issuing commands to a +// device. It covers the generic actuator capabilities used across the +// platform (device_types.capabilities: turn_on, turn_off, set_level) — no +// zone- or device-type-specific methods. A call here updates the device's +// desired state and dispatches the command over MQTT; it does not wait for +// the device to confirm execution (see the Device Shadow pattern in +// device-control-service's README). +service DeviceControl { + rpc TurnOn(TurnOnRequest) returns (CommandResult); + rpc TurnOff(TurnOffRequest) returns (CommandResult); + rpc SetLevel(SetLevelRequest) returns (CommandResult); +} + +message TurnOnRequest { + string device_id = 1; +} + +message TurnOffRequest { + string device_id = 1; +} + +message SetLevelRequest { + string device_id = 1; + double level = 2; +} + +message CommandResult { + bool success = 1; + string error = 2; +} diff --git a/proto/device_control/device_control_grpc.pb.go b/proto/device_control/device_control_grpc.pb.go new file mode 100644 index 0000000..0cf24aa --- /dev/null +++ b/proto/device_control/device_control_grpc.pb.go @@ -0,0 +1,213 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.2 +// - protoc v7.35.1 +// source: device_control/device_control.proto + +package device_control + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + DeviceControl_TurnOn_FullMethodName = "/device_control.DeviceControl/TurnOn" + DeviceControl_TurnOff_FullMethodName = "/device_control.DeviceControl/TurnOff" + DeviceControl_SetLevel_FullMethodName = "/device_control.DeviceControl/SetLevel" +) + +// DeviceControlClient is the client API for DeviceControl service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +// +// DeviceControl is the synchronous entry point for issuing commands to a +// device. It covers the generic actuator capabilities used across the +// platform (device_types.capabilities: turn_on, turn_off, set_level) — no +// zone- or device-type-specific methods. A call here updates the device's +// desired state and dispatches the command over MQTT; it does not wait for +// the device to confirm execution (see the Device Shadow pattern in +// device-control-service's README). +type DeviceControlClient interface { + TurnOn(ctx context.Context, in *TurnOnRequest, opts ...grpc.CallOption) (*CommandResult, error) + TurnOff(ctx context.Context, in *TurnOffRequest, opts ...grpc.CallOption) (*CommandResult, error) + SetLevel(ctx context.Context, in *SetLevelRequest, opts ...grpc.CallOption) (*CommandResult, error) +} + +type deviceControlClient struct { + cc grpc.ClientConnInterface +} + +func NewDeviceControlClient(cc grpc.ClientConnInterface) DeviceControlClient { + return &deviceControlClient{cc} +} + +func (c *deviceControlClient) TurnOn(ctx context.Context, in *TurnOnRequest, opts ...grpc.CallOption) (*CommandResult, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(CommandResult) + err := c.cc.Invoke(ctx, DeviceControl_TurnOn_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *deviceControlClient) TurnOff(ctx context.Context, in *TurnOffRequest, opts ...grpc.CallOption) (*CommandResult, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(CommandResult) + err := c.cc.Invoke(ctx, DeviceControl_TurnOff_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *deviceControlClient) SetLevel(ctx context.Context, in *SetLevelRequest, opts ...grpc.CallOption) (*CommandResult, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(CommandResult) + err := c.cc.Invoke(ctx, DeviceControl_SetLevel_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +// DeviceControlServer is the server API for DeviceControl service. +// All implementations must embed UnimplementedDeviceControlServer +// for forward compatibility. +// +// DeviceControl is the synchronous entry point for issuing commands to a +// device. It covers the generic actuator capabilities used across the +// platform (device_types.capabilities: turn_on, turn_off, set_level) — no +// zone- or device-type-specific methods. A call here updates the device's +// desired state and dispatches the command over MQTT; it does not wait for +// the device to confirm execution (see the Device Shadow pattern in +// device-control-service's README). +type DeviceControlServer interface { + TurnOn(context.Context, *TurnOnRequest) (*CommandResult, error) + TurnOff(context.Context, *TurnOffRequest) (*CommandResult, error) + SetLevel(context.Context, *SetLevelRequest) (*CommandResult, error) + mustEmbedUnimplementedDeviceControlServer() +} + +// UnimplementedDeviceControlServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedDeviceControlServer struct{} + +func (UnimplementedDeviceControlServer) TurnOn(context.Context, *TurnOnRequest) (*CommandResult, error) { + return nil, status.Error(codes.Unimplemented, "method TurnOn not implemented") +} +func (UnimplementedDeviceControlServer) TurnOff(context.Context, *TurnOffRequest) (*CommandResult, error) { + return nil, status.Error(codes.Unimplemented, "method TurnOff not implemented") +} +func (UnimplementedDeviceControlServer) SetLevel(context.Context, *SetLevelRequest) (*CommandResult, error) { + return nil, status.Error(codes.Unimplemented, "method SetLevel not implemented") +} +func (UnimplementedDeviceControlServer) mustEmbedUnimplementedDeviceControlServer() {} +func (UnimplementedDeviceControlServer) testEmbeddedByValue() {} + +// UnsafeDeviceControlServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to DeviceControlServer will +// result in compilation errors. +type UnsafeDeviceControlServer interface { + mustEmbedUnimplementedDeviceControlServer() +} + +func RegisterDeviceControlServer(s grpc.ServiceRegistrar, srv DeviceControlServer) { + // If the following call panics, it indicates UnimplementedDeviceControlServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&DeviceControl_ServiceDesc, srv) +} + +func _DeviceControl_TurnOn_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(TurnOnRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DeviceControlServer).TurnOn(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: DeviceControl_TurnOn_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DeviceControlServer).TurnOn(ctx, req.(*TurnOnRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _DeviceControl_TurnOff_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(TurnOffRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DeviceControlServer).TurnOff(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: DeviceControl_TurnOff_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DeviceControlServer).TurnOff(ctx, req.(*TurnOffRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _DeviceControl_SetLevel_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(SetLevelRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DeviceControlServer).SetLevel(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: DeviceControl_SetLevel_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DeviceControlServer).SetLevel(ctx, req.(*SetLevelRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// DeviceControl_ServiceDesc is the grpc.ServiceDesc for DeviceControl service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var DeviceControl_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "device_control.DeviceControl", + HandlerType: (*DeviceControlServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "TurnOn", + Handler: _DeviceControl_TurnOn_Handler, + }, + { + MethodName: "TurnOff", + Handler: _DeviceControl_TurnOff_Handler, + }, + { + MethodName: "SetLevel", + Handler: _DeviceControl_SetLevel_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "device_control/device_control.proto", +} diff --git a/proto/go.mod b/proto/go.mod new file mode 100644 index 0000000..482b548 --- /dev/null +++ b/proto/go.mod @@ -0,0 +1,15 @@ +module git.cactoz.su/cacto/home_automatization/proto + +go 1.25.0 + +require ( + google.golang.org/grpc v1.82.1 + google.golang.org/protobuf v1.36.11 +) + +require ( + golang.org/x/net v0.53.0 // indirect + golang.org/x/sys v0.43.0 // indirect + golang.org/x/text v0.36.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect +) diff --git a/proto/go.sum b/proto/go.sum new file mode 100644 index 0000000..7b006a0 --- /dev/null +++ b/proto/go.sum @@ -0,0 +1,38 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg= +golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= +google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/services/device-control-service/Dockerfile b/services/device-control-service/Dockerfile new file mode 100644 index 0000000..a6feb6f --- /dev/null +++ b/services/device-control-service/Dockerfile @@ -0,0 +1,15 @@ +# Build context is the repo root (not this directory) — device-control-service +# depends on the sibling proto/ Go module via a relative `replace` directive. +FROM golang:1.25-alpine AS build +WORKDIR /src +COPY proto/ ./proto/ +COPY services/device-control-service/ ./services/device-control-service/ +WORKDIR /src/services/device-control-service +RUN go mod download +RUN CGO_ENABLED=0 go build -o /out/device-control-service ./cmd/device-control-service + +FROM alpine:3.20 +RUN adduser -D -u 10001 app +COPY --from=build /out/device-control-service /usr/local/bin/device-control-service +USER app +ENTRYPOINT ["/usr/local/bin/device-control-service"] diff --git a/services/device-control-service/README.md b/services/device-control-service/README.md index 23aa8a6..c6e6805 100644 --- a/services/device-control-service/README.md +++ b/services/device-control-service/README.md @@ -1,20 +1,51 @@ # device-control-service (Go) -Статус: пока не реализован. +Статус: реализован (MVP). Зона ответственности: -- Предоставляет gRPC API (см. `proto/device_control.proto`) для отправки - команд устройствам — вызывается из Laravel (ручное управление) и из +- Предоставляет gRPC API (см. `proto/device_control/device_control.proto`: + `TurnOn`, `TurnOff`, `SetLevel` → `CommandResult`) для отправки команд + устройствам — вызывается из Laravel (ручное управление) и из rule-engine-service (действия по срабатыванию правил). - Публикует команду в MQTT-топик устройства (`devices/{device_id}/commands`) и слушает подтверждения (`devices/{device_id}/ack`). - Поддерживает паттерн Device Shadow в Redis: `desired_state` (чего хочет пользователь) против `reported_state` (что подтвердило устройство), плюс `status` и `last_seen`. -- Владеет циклом health-check (либо делегирует его health-check-service): - горутина-тикер переводит устройство в `offline`, когда `last_seen` - превышает таймаут, и генерирует событие для notification-service. +- Ведёт health-check горутиной-тикером: переводит устройство в `offline`, + когда `last_seen` превышает таймаут, и публикует событие в RabbitMQ + (`device.status_changed`) для будущего notification-service — как при + уходе в офлайн, так и при возврате online. Не входит в зону ответственности: решение о том, *когда* отправлять команду на основе показаний датчиков — это логика rule-engine-service. Этот сервис только исполняет команды и отслеживает состояние для той команды, что ему дали. + +## Архитектурные решения + +- **gRPC — не ждёт подтверждения устройства.** `CommandResult.success` + означает «desired_state обновлён и команда опубликована в MQTT», а не + «устройство подтвердило выполнение». Это соответствует Device Shadow: + мгновенный отклик в UI, даже если устройство офлайн или ответит с + задержкой. `reported_state` обновляется позже, асинхронно, по приходу ack. +- **last_seen обновляется и по телеметрии, и по ack.** Сервис подписан на + `devices/+/telemetry` только ради отметки «устройство живо» (сами данные + телеметрии парсит и хранит ingest-service) — иначе датчики без исходящих + команд (и, соответственно, без ack) никогда не считались бы online. +- **Нет зависимости от PostgreSQL.** Список известных устройств — это Redis + Set (`devices:known`), который пополняется по мере поступления + телеметрии/ack, а не выгружается из таблицы `devices`. Это удерживает + сервис в границах MQTT+Redis+RabbitMQ+gRPC, как и в диаграмме архитектуры + корневого README. + +## Запуск + +```bash +cd services/device-control-service +go test ./... +go build ./cmd/device-control-service +``` + +Конфигурация — через переменные окружения (секция device-control-service в +корневом `.env.example`). `proto/` — отдельный Go-модуль, подключается через +`replace` в `go.mod` на относительный путь `../../proto`. diff --git a/services/device-control-service/cmd/device-control-service/main.go b/services/device-control-service/cmd/device-control-service/main.go new file mode 100644 index 0000000..70e99d1 --- /dev/null +++ b/services/device-control-service/cmd/device-control-service/main.go @@ -0,0 +1,171 @@ +// Command device-control-service exposes a gRPC API for issuing device +// commands, dispatches them over MQTT, maintains the Device Shadow in +// Redis, and runs a health-check loop that flips stale devices offline. +package main + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "google.golang.org/grpc" + + devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control" + "git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/config" + "git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/healthcheck" + "git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/mqttclient" + "git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/rabbitmq" + "git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/server" + "git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/shadow" +) + +func main() { + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + + if err := run(logger); err != nil { + logger.Error("device-control-service exited with error", "error", err) + os.Exit(1) + } +} + +func run(logger *slog.Logger) error { + cfg, err := config.Load() + if err != nil { + return err + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + shadowStore, err := shadow.Connect(ctx, cfg.RedisAddr, cfg.RedisPassword, cfg.RedisDB) + cancel() + if err != nil { + return err + } + defer shadowStore.Close() + + publisher, err := rabbitmq.Connect(cfg.RabbitMQURL) + if err != nil { + return err + } + defer publisher.Close() + + mqttClient, err := mqttclient.Connect(mqttclient.Config{ + BrokerURL: cfg.MQTTBrokerURL, + ClientID: cfg.MQTTClientID, + }, logger) + if err != nil { + return err + } + defer mqttClient.Close() + + if err := mqttClient.Subscribe("devices/+/telemetry", 1, telemetryHandler(shadowStore, publisher, logger)); err != nil { + return fmt.Errorf("subscribe telemetry: %w", err) + } + if err := mqttClient.Subscribe("devices/+/ack", 1, ackHandler(shadowStore, publisher, logger)); err != nil { + return fmt.Errorf("subscribe ack: %w", err) + } + + checker := healthcheck.New(shadowStore, cfg.HealthCheckTimeout, cfg.HealthCheckInterval, func(deviceID string) { + pubCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := publisher.PublishStatusChanged(pubCtx, deviceID, shadow.StatusOffline); err != nil { + logger.Error("publish offline event failed", "device_id", deviceID, "error", err) + } + }, logger) + checker.Start() + defer checker.Stop() + + lis, err := net.Listen("tcp", fmt.Sprintf(":%d", cfg.GRPCPort)) + if err != nil { + return fmt.Errorf("listen on grpc port %d: %w", cfg.GRPCPort, err) + } + grpcServer := grpc.NewServer() + devicecontrol.RegisterDeviceControlServer(grpcServer, server.New(shadowStore, mqttClient, logger)) + + go func() { + logger.Info("device-control-service started", "grpc_port", cfg.GRPCPort) + if err := grpcServer.Serve(lis); err != nil { + logger.Error("grpc server stopped", "error", err) + } + }() + + stop := make(chan os.Signal, 1) + signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM) + <-stop + + logger.Info("shutting down") + grpcServer.GracefulStop() + + return nil +} + +// deviceIDFromTopic extracts {device_id} from "devices/{device_id}/". +func deviceIDFromTopic(topic string) (string, bool) { + parts := strings.Split(topic, "/") + if len(parts) != 3 || parts[0] != "devices" || parts[1] == "" { + return "", false + } + return parts[1], true +} + +// telemetryHandler only tracks liveness: ingest-service owns storing and +// interpreting telemetry, this service just needs to know the device is alive. +func telemetryHandler(store *shadow.Store, publisher *rabbitmq.Publisher, logger *slog.Logger) mqttclient.Handler { + return func(topic string, _ []byte, _ time.Time) { + deviceID, ok := deviceIDFromTopic(topic) + if !ok { + logger.Warn("dropping telemetry from unparseable topic", "topic", topic) + return + } + touch(context.Background(), store, publisher, deviceID, logger) + } +} + +type ackPayload struct { + State map[string]any `json:"state"` +} + +func ackHandler(store *shadow.Store, publisher *rabbitmq.Publisher, logger *slog.Logger) mqttclient.Handler { + return func(topic string, payload []byte, _ time.Time) { + deviceID, ok := deviceIDFromTopic(topic) + if !ok { + logger.Warn("dropping ack from unparseable topic", "topic", topic) + return + } + + var ack ackPayload + if err := json.Unmarshal(payload, &ack); err != nil { + logger.Warn("dropping invalid ack payload", "device_id", deviceID, "error", err) + return + } + + ctx := context.Background() + if len(ack.State) > 0 { + if err := store.PatchReportedState(ctx, deviceID, ack.State); err != nil { + logger.Error("patch reported state failed", "device_id", deviceID, "error", err) + } + } + touch(ctx, store, publisher, deviceID, logger) + } +} + +func touch(ctx context.Context, store *shadow.Store, publisher *rabbitmq.Publisher, deviceID string, logger *slog.Logger) { + becameOnline, err := store.Touch(ctx, deviceID) + if err != nil { + logger.Error("touch device failed", "device_id", deviceID, "error", err) + return + } + if becameOnline { + pubCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + if err := publisher.PublishStatusChanged(pubCtx, deviceID, shadow.StatusOnline); err != nil { + logger.Error("publish online event failed", "device_id", deviceID, "error", err) + } + } +} diff --git a/services/device-control-service/go.mod b/services/device-control-service/go.mod new file mode 100644 index 0000000..b102eca --- /dev/null +++ b/services/device-control-service/go.mod @@ -0,0 +1,27 @@ +module git.cactoz.su/cacto/home_automatization/services/device-control-service + +go 1.25.0 + +require ( + git.cactoz.su/cacto/home_automatization/proto v0.0.0 + github.com/alicebob/miniredis/v2 v2.38.0 + github.com/eclipse/paho.mqtt.golang v1.5.1 + github.com/rabbitmq/amqp091-go v1.13.0 + github.com/redis/go-redis/v9 v9.21.0 + google.golang.org/grpc v1.82.1 +) + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/gorilla/websocket v1.5.3 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/net v0.53.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.43.0 // indirect + golang.org/x/text v0.36.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect + google.golang.org/protobuf v1.36.11 // indirect +) + +replace git.cactoz.su/cacto/home_automatization/proto => ../../proto diff --git a/services/device-control-service/go.sum b/services/device-control-service/go.sum new file mode 100644 index 0000000..1ec4e21 --- /dev/null +++ b/services/device-control-service/go.sum @@ -0,0 +1,70 @@ +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE= +github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA= +github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E= +github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg= +golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= +google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/services/device-control-service/internal/config/config.go b/services/device-control-service/internal/config/config.go new file mode 100644 index 0000000..8826e00 --- /dev/null +++ b/services/device-control-service/internal/config/config.go @@ -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 +} diff --git a/services/device-control-service/internal/healthcheck/checker.go b/services/device-control-service/internal/healthcheck/checker.go new file mode 100644 index 0000000..3c2ed67 --- /dev/null +++ b/services/device-control-service/internal/healthcheck/checker.go @@ -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) + } + } + } +} diff --git a/services/device-control-service/internal/healthcheck/checker_test.go b/services/device-control-service/internal/healthcheck/checker_test.go new file mode 100644 index 0000000..1ad9ec8 --- /dev/null +++ b/services/device-control-service/internal/healthcheck/checker_test.go @@ -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 +} diff --git a/services/device-control-service/internal/mqttclient/client.go b/services/device-control-service/internal/mqttclient/client.go new file mode 100644 index 0000000..b974a13 --- /dev/null +++ b/services/device-control-service/internal/mqttclient/client.go @@ -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) +} diff --git a/services/device-control-service/internal/rabbitmq/publisher.go b/services/device-control-service/internal/rabbitmq/publisher.go new file mode 100644 index 0000000..7916ef2 --- /dev/null +++ b/services/device-control-service/internal/rabbitmq/publisher.go @@ -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 +} diff --git a/services/device-control-service/internal/server/server.go b/services/device-control-service/internal/server/server.go new file mode 100644 index 0000000..4c7e3be --- /dev/null +++ b/services/device-control-service/internal/server/server.go @@ -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 +} diff --git a/services/device-control-service/internal/server/server_test.go b/services/device-control-service/internal/server/server_test.go new file mode 100644 index 0000000..ae5f424 --- /dev/null +++ b/services/device-control-service/internal/server/server_test.go @@ -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") + } +} diff --git a/services/device-control-service/internal/shadow/store.go b/services/device-control-service/internal/shadow/store.go new file mode 100644 index 0000000..ba1c536 --- /dev/null +++ b/services/device-control-service/internal/shadow/store.go @@ -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 +} diff --git a/services/device-control-service/internal/shadow/store_test.go b/services/device-control-service/internal/shadow/store_test.go new file mode 100644 index 0000000..b8f573e --- /dev/null +++ b/services/device-control-service/internal/shadow/store_test.go @@ -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") + } +}