Реализация rule-engine-service: RabbitMQ → правила из PostgreSQL → gRPC

Слушает telemetry.new_reading (ручной ack/nack, реквизишн только при
транспортных ошибках gRPC), кэширует активные automation_rules в памяти
с периодическим обновлением из PostgreSQL (JOIN с devices — резолвит
внутренние ID в external_id, которым оперируют MQTT/Redis/gRPC). Условия
правил (>,<,>=,<=,=,!=) оцениваются обобщённо, без привязки к конкретным
типам устройств. При срабатывании вызывает device-control-service по gRPC
и публикует automation.rule_triggered в RabbitMQ.

Проверено сквозным тестом через docker compose на полном пайплайне:
mosquitto_pub → ingest-service (ClickHouse + telemetry.new_reading) →
rule-engine-service (совпадение правила) → device-control-service (gRPC →
Redis desired_state + MQTT-команда) → automation.rule_triggered. Показание
ниже порога проверено отдельно — правило корректно не срабатывает.

.env.example

@@ -43,6 +43,10 @@ INGEST_BATCH_MAX_SIZE=500
 INGEST_BATCH_FLUSH_INTERVAL=5s
 INGEST_BATCH_FLUSH_TIMEOUT=10s
 
+# --- rule-engine-service ---
+DEVICE_CONTROL_HOST=device-control-service
+RULE_ENGINE_CACHE_REFRESH_INTERVAL=15s
+
 # --- Laravel app (stage 2) ---
 APP_KEY=
 APP_URL=http://localhost:8000

docker-compose.yml

@@ -147,6 +147,34 @@ services:
       - home-automation
     restart: unless-stopped
 
-  # Remaining app services (rule-engine-service, health-check-service,
+  rule-engine-service:
-  # esp32-emulator, laravel-app) are added here as they get implemented —
+    build:
-  # see services/*/README.md for the plan for each one.
+      context: .
+      dockerfile: services/rule-engine-service/Dockerfile
+    environment:
+      POSTGRES_HOST: postgres
+      POSTGRES_PORT: 5432
+      POSTGRES_DB: ${POSTGRES_DB}
+      POSTGRES_USER: ${POSTGRES_USER}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+      DEVICE_CONTROL_HOST: device-control-service
+      DEVICE_CONTROL_GRPC_PORT: 50051
+      RABBITMQ_HOST: rabbitmq
+      RABBITMQ_PORT: 5672
+      RABBITMQ_USER: ${RABBITMQ_USER}
+      RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD}
+      RULE_ENGINE_CACHE_REFRESH_INTERVAL: ${RULE_ENGINE_CACHE_REFRESH_INTERVAL}
+    depends_on:
+      postgres:
+        condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
+      device-control-service:
+        condition: service_started
+    networks:
+      - home-automation
+    restart: unless-stopped
+
+  # Remaining app services (health-check-service, esp32-emulator,
+  # laravel-app) are added here as they get implemented — see
+  # services/*/README.md for the plan for each one.

services/rule-engine-service/Dockerfile

@@ -0,0 +1,15 @@
+# Build context is the repo root (not this directory) — rule-engine-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/rule-engine-service/ ./services/rule-engine-service/
+WORKDIR /src/services/rule-engine-service
+RUN go mod download
+RUN CGO_ENABLED=0 go build -o /out/rule-engine-service ./cmd/rule-engine-service
+
+FROM alpine:3.20
+RUN adduser -D -u 10001 app
+COPY --from=build /out/rule-engine-service /usr/local/bin/rule-engine-service
+USER app
+ENTRYPOINT ["/usr/local/bin/rule-engine-service"]

services/rule-engine-service/README.md

@@ -1,18 +1,52 @@
 # rule-engine-service (Go)
 
-Статус: пока не реализован.
+Статус: реализован (MVP).
 
 Зона ответственности:
-- Потребляет события «новое показание» из RabbitMQ (асинхронно — здесь
+- Потребляет события «новое показание» из RabbitMQ (очередь
-  надёжность доставки важнее задержки).
+  `telemetry.new_reading`, публикуемая ingest-service) — асинхронно, с ручным
-- Загружает активные `automation_rules` для нужной зоны/устройства из
+  ack/nack: надёжность доставки здесь важнее задержки.
-  PostgreSQL, кэшируя в памяти/Redis, чтобы не ходить в БД на каждое событие.
+- Загружает активные `automation_rules` из PostgreSQL в кэш в памяти
+  (обновляется по тикеру), чтобы не ходить в БД на каждое событие.
 - Оценивает условия правил обобщённо: `sensor_type X operator value` против
   показания — никаких зашитых названий датчиков/устройств, всё берётся из
   строки `automation_rules`.
 - При срабатывании: вызывает device-control-service по gRPC (нужен быстрый
-  результат успех/неудача) и публикует событие в notification-service через
+  результат успех/неудача) и публикует событие `automation.rule_triggered` в
-  RabbitMQ (не критично к задержке).
+  RabbitMQ для будущего notification-service (не критично к задержке).
 
 Не входит в зону ответственности: прямое взаимодействие с MQTT или Redis —
 изменения состояния устройства всегда идут через gRPC API device-control-service.
+
+## Архитектурные решения
+
+- **automation_rules хранит внутренние ID Postgres, а не внешние device_id.**
+  `condition_source_device_id`/`target_device_id` — это FK на `devices.id`
+  (bigserial), а MQTT/Redis/gRPC везде оперируют `devices.external_id`
+  (строка). Поэтому кэш правил грузится JOIN'ом `automation_rules` с
+  `devices` (дважды — для source и target), резолвя оба ID в external_id
+  один раз при загрузке, а не на каждое событие.
+- **Retry — только на транспортных ошибках.** Если gRPC-вызов
+  device-control-service не удался физически (сеть, сервис недоступен) —
+  событие из RabbitMQ nack'ается с `requeue=true`, всё правило
+  переигрывается позже. Если устройство само отклонило команду
+  (`CommandResult.success=false`, например неизвестный device_id) — это
+  финальный исход, ack, повторов не будет. Осознанное упрощение: если из
+  нескольких правил на одно показание одно не удалось из-за сети, при
+  повторной доставке переиграются ВСЕ правила читающие это показание, включая
+  уже успешно сработавшие — дедупликация не реализована, для MVP это
+  приемлемо (команды идемпотентны: turn_on дважды — не проблема).
+- **Кэш правил — только для чтения активных правил**, никакой записи назад в
+  Postgres. CRUD правил — это зона ответственности Laravel (этап 2).
+
+## Запуск
+
+```bash
+cd services/rule-engine-service
+go test ./...
+go build ./cmd/rule-engine-service
+```
+
+Конфигурация — через переменные окружения (секция rule-engine-service в
+корневом `.env.example`). Как и device-control-service, использует
+`proto/device_control` через `replace` на `../../proto` в `go.mod`.

services/rule-engine-service/cmd/rule-engine-service/main.go

@@ -0,0 +1,88 @@
+// Command rule-engine-service consumes telemetry readings from RabbitMQ,
+// evaluates them against PostgreSQL-backed automation rules, and dispatches
+// matching actions to device-control-service over gRPC.
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/config"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/engine"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rabbitmq"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rulecache"
+)
+
+func main() {
+	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
+
+	if err := run(logger); err != nil {
+		logger.Error("rule-engine-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, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	pgPool, err := pgxpool.New(ctx, cfg.PostgresDSN)
+	if err != nil {
+		return err
+	}
+	defer pgPool.Close()
+
+	dcClient, err := devicecontrolclient.Connect(cfg.DeviceControlGRPCTarget)
+	if err != nil {
+		return err
+	}
+	defer dcClient.Close()
+
+	mq, err := rabbitmq.Connect(cfg.RabbitMQURL)
+	if err != nil {
+		return err
+	}
+	defer mq.Close()
+
+	cache := rulecache.New(rulecache.NewPostgresFetcher(pgPool), cfg.RuleCacheRefreshInterval, logger)
+	if err := cache.Start(ctx); err != nil {
+		return err
+	}
+	defer cache.Stop()
+
+	eng := engine.New(cache, dcClient, mq, logger)
+
+	logger.Info("rule-engine-service started")
+
+	err = mq.ConsumeReadings(ctx, func(ctx context.Context, event rabbitmq.ReadingEvent) rabbitmq.HandleResult {
+		reading := engine.Reading{
+			DeviceID:   event.DeviceID,
+			ZoneID:     event.ZoneID,
+			SensorType: event.SensorType,
+			Value:      event.Value,
+			RecordedAt: event.RecordedAt,
+		}
+		if err := eng.HandleReading(ctx, reading); err != nil {
+			logger.Error("handling reading failed, will retry", "device_id", event.DeviceID, "error", err)
+			return rabbitmq.NackRequeue
+		}
+		return rabbitmq.Ack
+	})
+	if err != nil {
+		return err
+	}
+
+	logger.Info("shutting down")
+	return nil
+}

services/rule-engine-service/go.mod

@@ -0,0 +1,24 @@
+module git.cactoz.su/cacto/home_automatization/services/rule-engine-service
+
+go 1.25.0
+
+require (
+	git.cactoz.su/cacto/home_automatization/proto v0.0.0-00010101000000-000000000000
+	github.com/jackc/pgx/v5 v5.10.0
+	github.com/rabbitmq/amqp091-go v1.13.0
+	google.golang.org/grpc v1.82.1
+)
+
+require (
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // 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

services/rule-engine-service/go.sum

@@ -0,0 +1,66 @@
+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.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+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/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/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+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/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+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/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=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

services/rule-engine-service/internal/config/config.go

@@ -0,0 +1,70 @@
+// Package config loads rule-engine-service settings from environment
+// variables, matching the names used in the repo-root .env.example.
+package config
+
+import (
+	"fmt"
+	"os"
+	"time"
+)
+
+type Config struct {
+	PostgresDSN string
+
+	DeviceControlGRPCTarget string
+
+	RabbitMQURL string
+
+	RuleCacheRefreshInterval time.Duration
+}
+
+func Load() (Config, error) {
+	cfg := Config{
+		PostgresDSN: fmt.Sprintf("postgres://%s:%s@%s:%s/%s",
+			getEnv("POSTGRES_USER", "home_automation"),
+			getEnv("POSTGRES_PASSWORD", ""),
+			getEnv("POSTGRES_HOST", "localhost"),
+			getEnv("POSTGRES_PORT", "5432"),
+			getEnv("POSTGRES_DB", "home_automation"),
+		),
+
+		DeviceControlGRPCTarget: fmt.Sprintf("%s:%s",
+			getEnv("DEVICE_CONTROL_HOST", "localhost"),
+			getEnv("DEVICE_CONTROL_GRPC_PORT", "50051"),
+		),
+
+		RabbitMQURL: fmt.Sprintf("amqp://%s:%s@%s:%s/",
+			getEnv("RABBITMQ_USER", "guest"),
+			getEnv("RABBITMQ_PASSWORD", "guest"),
+			getEnv("RABBITMQ_HOST", "localhost"),
+			getEnv("RABBITMQ_PORT", "5672"),
+		),
+	}
+
+	interval, err := getEnvDuration("RULE_ENGINE_CACHE_REFRESH_INTERVAL", 15*time.Second)
+	if err != nil {
+		return Config{}, err
+	}
+	cfg.RuleCacheRefreshInterval = interval
+
+	return cfg, nil
+}
+
+func getEnv(key, fallback string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return fallback
+}
+
+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
+}

services/rule-engine-service/internal/devicecontrolclient/client.go

@@ -0,0 +1,95 @@
+// Package devicecontrolclient wraps the generated DeviceControl gRPC client,
+// translating an automation_rules row's generic action_type/action_params
+// into the specific TurnOn/TurnOff/SetLevel RPC.
+package devicecontrolclient
+
+import (
+	"context"
+	"fmt"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
+)
+
+type Client struct {
+	conn   *grpc.ClientConn
+	client devicecontrol.DeviceControlClient
+}
+
+func Connect(target string) (*Client, error) {
+	conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, fmt.Errorf("connect to device-control-service at %s: %w", target, err)
+	}
+	return &Client{conn: conn, client: devicecontrol.NewDeviceControlClient(conn)}, nil
+}
+
+func (c *Client) Close() error {
+	return c.conn.Close()
+}
+
+// newWithClient builds a Client around an already-constructed gRPC client,
+// bypassing Connect. Used by tests to inject a fake devicecontrol.DeviceControlClient.
+func newWithClient(client devicecontrol.DeviceControlClient) *Client {
+	return &Client{client: client}
+}
+
+// Result mirrors devicecontrol.CommandResult: Success/Error come from the
+// device-control-service's business logic, not a transport-level failure.
+type Result struct {
+	Success bool
+	Error   string
+}
+
+// Dispatch issues actionType against deviceID, using params for actions that
+// need them (currently just set_level's "level"). The returned error is only
+// set for transport/protocol failures; a rejected command comes back as
+// Result{Success: false, Error: "..."}.
+func (c *Client) Dispatch(ctx context.Context, deviceID, actionType string, params map[string]any) (Result, error) {
+	switch actionType {
+	case "turn_on":
+		res, err := c.client.TurnOn(ctx, &devicecontrol.TurnOnRequest{DeviceId: deviceID})
+		if err != nil {
+			return Result{}, fmt.Errorf("turn_on rpc: %w", err)
+		}
+		return Result{Success: res.GetSuccess(), Error: res.GetError()}, nil
+
+	case "turn_off":
+		res, err := c.client.TurnOff(ctx, &devicecontrol.TurnOffRequest{DeviceId: deviceID})
+		if err != nil {
+			return Result{}, fmt.Errorf("turn_off rpc: %w", err)
+		}
+		return Result{Success: res.GetSuccess(), Error: res.GetError()}, nil
+
+	case "set_level":
+		level, ok := numericParam(params, "level")
+		if !ok {
+			return Result{}, fmt.Errorf("set_level requires a numeric %q param, got %+v", "level", params)
+		}
+		res, err := c.client.SetLevel(ctx, &devicecontrol.SetLevelRequest{DeviceId: deviceID, Level: level})
+		if err != nil {
+			return Result{}, fmt.Errorf("set_level rpc: %w", err)
+		}
+		return Result{Success: res.GetSuccess(), Error: res.GetError()}, nil
+
+	default:
+		return Result{}, fmt.Errorf("unknown action_type %q", actionType)
+	}
+}
+
+func numericParam(params map[string]any, key string) (float64, bool) {
+	v, ok := params[key]
+	if !ok {
+		return 0, false
+	}
+	switch n := v.(type) {
+	case float64:
+		return n, true
+	case int:
+		return float64(n), true
+	default:
+		return 0, false
+	}
+}

services/rule-engine-service/internal/devicecontrolclient/client_test.go

@@ -0,0 +1,94 @@
+package devicecontrolclient
+
+import (
+	"context"
+	"testing"
+
+	"google.golang.org/grpc"
+
+	devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
+)
+
+type fakeGRPCClient struct {
+	lastTurnOn   *devicecontrol.TurnOnRequest
+	lastTurnOff  *devicecontrol.TurnOffRequest
+	lastSetLevel *devicecontrol.SetLevelRequest
+	result       *devicecontrol.CommandResult
+	err          error
+}
+
+func (f *fakeGRPCClient) TurnOn(_ context.Context, in *devicecontrol.TurnOnRequest, _ ...grpc.CallOption) (*devicecontrol.CommandResult, error) {
+	f.lastTurnOn = in
+	return f.result, f.err
+}
+
+func (f *fakeGRPCClient) TurnOff(_ context.Context, in *devicecontrol.TurnOffRequest, _ ...grpc.CallOption) (*devicecontrol.CommandResult, error) {
+	f.lastTurnOff = in
+	return f.result, f.err
+}
+
+func (f *fakeGRPCClient) SetLevel(_ context.Context, in *devicecontrol.SetLevelRequest, _ ...grpc.CallOption) (*devicecontrol.CommandResult, error) {
+	f.lastSetLevel = in
+	return f.result, f.err
+}
+
+func TestDispatch_TurnOn(t *testing.T) {
+	fake := &fakeGRPCClient{result: &devicecontrol.CommandResult{Success: true}}
+	c := newWithClient(fake)
+
+	res, err := c.Dispatch(context.Background(), "fan-1", "turn_on", nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !res.Success {
+		t.Fatalf("got %+v, want success", res)
+	}
+	if fake.lastTurnOn == nil || fake.lastTurnOn.DeviceId != "fan-1" {
+		t.Fatalf("got TurnOn request %+v, want device_id=fan-1", fake.lastTurnOn)
+	}
+}
+
+func TestDispatch_SetLevel(t *testing.T) {
+	fake := &fakeGRPCClient{result: &devicecontrol.CommandResult{Success: true}}
+	c := newWithClient(fake)
+
+	res, err := c.Dispatch(context.Background(), "light-1", "set_level", map[string]any{"level": 42.5})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !res.Success {
+		t.Fatalf("got %+v, want success", res)
+	}
+	if fake.lastSetLevel == nil || fake.lastSetLevel.Level != 42.5 {
+		t.Fatalf("got SetLevel request %+v, want level=42.5", fake.lastSetLevel)
+	}
+}
+
+func TestDispatch_SetLevel_MissingParam(t *testing.T) {
+	c := newWithClient(&fakeGRPCClient{})
+
+	if _, err := c.Dispatch(context.Background(), "light-1", "set_level", nil); err == nil {
+		t.Fatal("expected error for missing level param")
+	}
+}
+
+func TestDispatch_UnknownActionType(t *testing.T) {
+	c := newWithClient(&fakeGRPCClient{})
+
+	if _, err := c.Dispatch(context.Background(), "d1", "dance", nil); err == nil {
+		t.Fatal("expected error for unknown action_type")
+	}
+}
+
+func TestDispatch_BusinessFailurePassesThrough(t *testing.T) {
+	fake := &fakeGRPCClient{result: &devicecontrol.CommandResult{Success: false, Error: "device offline"}}
+	c := newWithClient(fake)
+
+	res, err := c.Dispatch(context.Background(), "fan-1", "turn_off", nil)
+	if err != nil {
+		t.Fatalf("unexpected transport error: %v", err)
+	}
+	if res.Success || res.Error != "device offline" {
+		t.Fatalf("got %+v, want business failure passed through", res)
+	}
+}

services/rule-engine-service/internal/engine/engine.go

@@ -0,0 +1,96 @@
+// Package engine orchestrates the core rule-engine loop: given one sensor
+// reading, find the rules that watch it, evaluate their conditions, and
+// dispatch the matching actions. It knows nothing about RabbitMQ, gRPC, or
+// PostgreSQL directly — those are injected as narrow interfaces so this
+// package is testable without any of them running.
+package engine
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
+)
+
+// Reading is the generic sensor reading the engine reacts to — no
+// growbox/device-type-specific fields, matching the platform's abstractions.
+type Reading struct {
+	DeviceID   string
+	ZoneID     string
+	SensorType string
+	Value      float64
+	RecordedAt time.Time
+}
+
+// RuleSource is the subset of rulecache.Cache the engine needs.
+type RuleSource interface {
+	MatchingRules(deviceID, sensorType string) []rules.Rule
+}
+
+// Dispatcher is the subset of devicecontrolclient.Client the engine needs.
+type Dispatcher interface {
+	Dispatch(ctx context.Context, deviceID, actionType string, params map[string]any) (devicecontrolclient.Result, error)
+}
+
+// TriggerPublisher is the subset of rabbitmq.Client the engine needs.
+type TriggerPublisher interface {
+	PublishRuleTriggered(ctx context.Context, ruleID, zoneID int64, deviceID, actionType string, success bool, errMsg string) error
+}
+
+type Engine struct {
+	rules      RuleSource
+	dispatcher Dispatcher
+	publisher  TriggerPublisher
+	logger     *slog.Logger
+}
+
+func New(ruleSource RuleSource, dispatcher Dispatcher, publisher TriggerPublisher, logger *slog.Logger) *Engine {
+	return &Engine{rules: ruleSource, dispatcher: dispatcher, publisher: publisher, logger: logger}
+}
+
+// HandleReading evaluates every rule watching (r.DeviceID, r.SensorType) and
+// dispatches the ones whose condition matches. It returns a non-nil error
+// only when a dispatch failed at the transport level (e.g.
+// device-control-service unreachable) — that's the one case worth retrying
+// the whole reading for. A rule the device itself rejected (bad device_id,
+// unsupported action) is terminal and does not cause a retry.
+func (e *Engine) HandleReading(ctx context.Context, r Reading) error {
+	matched := e.rules.MatchingRules(r.DeviceID, r.SensorType)
+
+	var firstTransportErr error
+	for _, rule := range matched {
+		matchesCondition, err := rules.Evaluate(rule.ConditionOperator, r.Value, rule.ConditionValue)
+		if err != nil {
+			e.logger.Error("skipping rule with invalid condition", "rule_id", rule.ID, "error", err)
+			continue
+		}
+		if !matchesCondition {
+			continue
+		}
+
+		result, dispatchErr := e.dispatcher.Dispatch(ctx, rule.TargetDeviceID, rule.ActionType, rule.ActionParams)
+
+		success := dispatchErr == nil && result.Success
+		errMsg := result.Error
+		switch {
+		case dispatchErr != nil:
+			errMsg = dispatchErr.Error()
+			e.logger.Error("dispatch action failed", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "error", dispatchErr)
+			if firstTransportErr == nil {
+				firstTransportErr = dispatchErr
+			}
+		case !result.Success:
+			e.logger.Warn("device rejected command", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "error", result.Error)
+		default:
+			e.logger.Info("rule triggered", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "action_type", rule.ActionType)
+		}
+
+		if pubErr := e.publisher.PublishRuleTriggered(ctx, rule.ID, rule.ZoneID, rule.TargetDeviceID, rule.ActionType, success, errMsg); pubErr != nil {
+			e.logger.Error("publish rule_triggered event failed", "rule_id", rule.ID, "error", pubErr)
+		}
+	}
+
+	return firstTransportErr
+}

services/rule-engine-service/internal/engine/engine_test.go

@@ -0,0 +1,156 @@
+package engine
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"testing"
+
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
+)
+
+type fakeRuleSource struct {
+	rules []rules.Rule
+}
+
+func (f *fakeRuleSource) MatchingRules(_, _ string) []rules.Rule {
+	return f.rules
+}
+
+type dispatchCall struct {
+	deviceID   string
+	actionType string
+	params     map[string]any
+}
+
+type fakeDispatcher struct {
+	calls  []dispatchCall
+	result devicecontrolclient.Result
+	err    error
+}
+
+func (f *fakeDispatcher) Dispatch(_ context.Context, deviceID, actionType string, params map[string]any) (devicecontrolclient.Result, error) {
+	f.calls = append(f.calls, dispatchCall{deviceID: deviceID, actionType: actionType, params: params})
+	return f.result, f.err
+}
+
+type triggeredEvent struct {
+	ruleID     int64
+	deviceID   string
+	actionType string
+	success    bool
+	errMsg     string
+}
+
+type fakePublisher struct {
+	events []triggeredEvent
+}
+
+func (f *fakePublisher) PublishRuleTriggered(_ context.Context, ruleID, _ int64, deviceID, actionType string, success bool, errMsg string) error {
+	f.events = append(f.events, triggeredEvent{ruleID: ruleID, deviceID: deviceID, actionType: actionType, success: success, errMsg: errMsg})
+	return nil
+}
+
+func discardLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func TestHandleReading_NoMatchingRules(t *testing.T) {
+	dispatcher := &fakeDispatcher{}
+	publisher := &fakePublisher{}
+	e := New(&fakeRuleSource{}, dispatcher, publisher, discardLogger())
+
+	if err := e.HandleReading(context.Background(), Reading{DeviceID: "sensor-1", SensorType: "temperature", Value: 30}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(dispatcher.calls) != 0 || len(publisher.events) != 0 {
+		t.Fatalf("expected no dispatch/publish, got %+v / %+v", dispatcher.calls, publisher.events)
+	}
+}
+
+func TestHandleReading_ConditionNotMet(t *testing.T) {
+	source := &fakeRuleSource{rules: []rules.Rule{
+		{ID: 1, ConditionOperator: ">", ConditionValue: 28, TargetDeviceID: "fan-1", ActionType: "turn_on"},
+	}}
+	dispatcher := &fakeDispatcher{}
+	e := New(source, dispatcher, &fakePublisher{}, discardLogger())
+
+	if err := e.HandleReading(context.Background(), Reading{Value: 20}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(dispatcher.calls) != 0 {
+		t.Fatalf("expected no dispatch when condition not met, got %+v", dispatcher.calls)
+	}
+}
+
+func TestHandleReading_ConditionMet_DispatchesAndPublishes(t *testing.T) {
+	source := &fakeRuleSource{rules: []rules.Rule{
+		{ID: 1, ZoneID: 7, ConditionOperator: ">", ConditionValue: 28, TargetDeviceID: "fan-1", ActionType: "turn_on"},
+	}}
+	dispatcher := &fakeDispatcher{result: devicecontrolclient.Result{Success: true}}
+	publisher := &fakePublisher{}
+	e := New(source, dispatcher, publisher, discardLogger())
+
+	if err := e.HandleReading(context.Background(), Reading{Value: 30}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if len(dispatcher.calls) != 1 || dispatcher.calls[0].deviceID != "fan-1" || dispatcher.calls[0].actionType != "turn_on" {
+		t.Fatalf("got dispatch calls %+v, want one turn_on for fan-1", dispatcher.calls)
+	}
+	if len(publisher.events) != 1 || !publisher.events[0].success {
+		t.Fatalf("got publish events %+v, want one success event", publisher.events)
+	}
+}
+
+func TestHandleReading_BusinessRejection_NotRetried(t *testing.T) {
+	source := &fakeRuleSource{rules: []rules.Rule{
+		{ID: 1, ConditionOperator: ">", ConditionValue: 28, TargetDeviceID: "fan-1", ActionType: "turn_on"},
+	}}
+	dispatcher := &fakeDispatcher{result: devicecontrolclient.Result{Success: false, Error: "device offline"}}
+	publisher := &fakePublisher{}
+	e := New(source, dispatcher, publisher, discardLogger())
+
+	err := e.HandleReading(context.Background(), Reading{Value: 30})
+	if err != nil {
+		t.Fatalf("business rejection should not be reported as a retryable error, got %v", err)
+	}
+	if len(publisher.events) != 1 || publisher.events[0].success || publisher.events[0].errMsg != "device offline" {
+		t.Fatalf("got publish events %+v, want one failed event with device offline", publisher.events)
+	}
+}
+
+func TestHandleReading_TransportError_IsRetryable(t *testing.T) {
+	source := &fakeRuleSource{rules: []rules.Rule{
+		{ID: 1, ConditionOperator: ">", ConditionValue: 28, TargetDeviceID: "fan-1", ActionType: "turn_on"},
+	}}
+	dispatcher := &fakeDispatcher{err: errors.New("connection refused")}
+	publisher := &fakePublisher{}
+	e := New(source, dispatcher, publisher, discardLogger())
+
+	err := e.HandleReading(context.Background(), Reading{Value: 30})
+	if err == nil {
+		t.Fatal("expected a retryable error when dispatch fails at the transport level")
+	}
+	if len(publisher.events) != 1 || publisher.events[0].success {
+		t.Fatalf("got publish events %+v, want one failed event recorded even on transport error", publisher.events)
+	}
+}
+
+func TestHandleReading_InvalidOperatorSkipsRuleButContinues(t *testing.T) {
+	source := &fakeRuleSource{rules: []rules.Rule{
+		{ID: 1, ConditionOperator: "~=", ConditionValue: 28, TargetDeviceID: "bad-rule"},
+		{ID: 2, ConditionOperator: ">", ConditionValue: 28, TargetDeviceID: "fan-1", ActionType: "turn_on"},
+	}}
+	dispatcher := &fakeDispatcher{result: devicecontrolclient.Result{Success: true}}
+	e := New(source, dispatcher, &fakePublisher{}, discardLogger())
+
+	if err := e.HandleReading(context.Background(), Reading{Value: 30}); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(dispatcher.calls) != 1 || dispatcher.calls[0].deviceID != "fan-1" {
+		t.Fatalf("got dispatch calls %+v, want only the valid rule dispatched", dispatcher.calls)
+	}
+}

services/rule-engine-service/internal/rabbitmq/client.go

@@ -0,0 +1,62 @@
+// Package rabbitmq consumes ingest-service's "new reading" events and
+// publishes rule-trigger events for notification-service. One connection is
+// shared, with separate channels for consuming and publishing, since that's
+// the amqp best practice (a channel is not meant to be used concurrently for
+// unrelated flows).
+package rabbitmq
+
+import (
+	"fmt"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+const (
+	NewReadingQueue    = "telemetry.new_reading" // declared by ingest-service; declared here too, idempotently
+	RuleTriggeredQueue = "automation.rule_triggered"
+)
+
+type Client struct {
+	conn      *amqp.Connection
+	consumeCh *amqp.Channel
+	publishCh *amqp.Channel
+}
+
+func Connect(url string) (*Client, error) {
+	conn, err := amqp.Dial(url)
+	if err != nil {
+		return nil, fmt.Errorf("dial rabbitmq: %w", err)
+	}
+
+	consumeCh, err := conn.Channel()
+	if err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("open consume channel: %w", err)
+	}
+	// Process one reading at a time: simple, demonstrable backpressure —
+	// the next delivery isn't sent until the current one is acked/nacked.
+	if err := consumeCh.Qos(1, 0, false); err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("set qos: %w", err)
+	}
+	if _, err := consumeCh.QueueDeclare(NewReadingQueue, true, false, false, false, nil); err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("declare queue %s: %w", NewReadingQueue, err)
+	}
+
+	publishCh, err := conn.Channel()
+	if err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("open publish channel: %w", err)
+	}
+	if _, err := publishCh.QueueDeclare(RuleTriggeredQueue, true, false, false, false, nil); err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("declare queue %s: %w", RuleTriggeredQueue, err)
+	}
+
+	return &Client{conn: conn, consumeCh: consumeCh, publishCh: publishCh}, nil
+}
+
+func (c *Client) Close() error {
+	return c.conn.Close()
+}

services/rule-engine-service/internal/rabbitmq/consumer.go

@@ -0,0 +1,65 @@
+package rabbitmq
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// ReadingEvent mirrors ingest-service's newReadingEvent wire format.
+type ReadingEvent struct {
+	DeviceID   string    `json:"device_id"`
+	ZoneID     string    `json:"zone_id"`
+	SensorType string    `json:"sensor_type"`
+	Value      float64   `json:"value"`
+	RecordedAt time.Time `json:"recorded_at"`
+}
+
+// HandleResult tells ConsumeReadings how to settle a delivery.
+type HandleResult int
+
+const (
+	Ack         HandleResult = iota // processed successfully
+	NackRequeue                     // transient failure (e.g. gRPC unreachable) — try again later
+	NackDiscard                     // permanent failure (e.g. bad data) — retrying won't help
+)
+
+// Handler processes one reading event and decides how it should be settled.
+type Handler func(ctx context.Context, event ReadingEvent) HandleResult
+
+// ConsumeReadings blocks, delivering messages to handler, until ctx is
+// cancelled or the delivery channel closes (e.g. connection lost).
+func (c *Client) ConsumeReadings(ctx context.Context, handler Handler) error {
+	deliveries, err := c.consumeCh.ConsumeWithContext(ctx, NewReadingQueue, "", false, false, false, false, nil)
+	if err != nil {
+		return fmt.Errorf("consume %s: %w", NewReadingQueue, err)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case d, ok := <-deliveries:
+			if !ok {
+				return fmt.Errorf("delivery channel for %s closed", NewReadingQueue)
+			}
+
+			var event ReadingEvent
+			if err := json.Unmarshal(d.Body, &event); err != nil {
+				// Poison message: no amount of retrying fixes malformed JSON.
+				_ = d.Nack(false, false)
+				continue
+			}
+
+			switch handler(ctx, event) {
+			case Ack:
+				_ = d.Ack(false)
+			case NackRequeue:
+				_ = d.Nack(false, true)
+			case NackDiscard:
+				_ = d.Nack(false, false)
+			}
+		}
+	}
+}

services/rule-engine-service/internal/rabbitmq/publisher.go

@@ -0,0 +1,48 @@
+package rabbitmq
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type ruleTriggeredEvent struct {
+	RuleID     int64     `json:"rule_id"`
+	ZoneID     int64     `json:"zone_id"`
+	DeviceID   string    `json:"device_id"`
+	ActionType string    `json:"action_type"`
+	Success    bool      `json:"success"`
+	Error      string    `json:"error,omitempty"`
+	At         time.Time `json:"at"`
+}
+
+// PublishRuleTriggered emits one event per rule match+dispatch attempt, so
+// notification-service can (eventually) alert on it.
+func (c *Client) PublishRuleTriggered(ctx context.Context, ruleID, zoneID int64, deviceID, actionType string, success bool, errMsg string) error {
+	body, err := json.Marshal(ruleTriggeredEvent{
+		RuleID:     ruleID,
+		ZoneID:     zoneID,
+		DeviceID:   deviceID,
+		ActionType: actionType,
+		Success:    success,
+		Error:      errMsg,
+		At:         time.Now(),
+	})
+	if err != nil {
+		return fmt.Errorf("marshal event: %w", err)
+	}
+
+	err = c.publishCh.PublishWithContext(ctx, "", RuleTriggeredQueue, false, false, amqp.Publishing{
+		ContentType:  "application/json",
+		DeliveryMode: amqp.Persistent,
+		Timestamp:    time.Now(),
+		Body:         body,
+	})
+	if err != nil {
+		return fmt.Errorf("publish rule triggered event for rule %d: %w", ruleID, err)
+	}
+	return nil
+}

services/rule-engine-service/internal/rulecache/cache.go

@@ -0,0 +1,139 @@
+// Package rulecache keeps active automation rules in memory, refreshed
+// periodically from PostgreSQL, so the hot path (one Redis/gRPC-free lookup
+// per incoming reading) never needs a database round-trip.
+package rulecache
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"sync"
+	"time"
+
+	"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
+)
+
+// Row is one automation_rules record already joined against devices, so
+// SourceExternalID/TargetExternalID are the external device IDs used by
+// MQTT/Redis/gRPC — not PostgreSQL's internal bigserial IDs.
+type Row struct {
+	RuleID              int64
+	ZoneID              int64
+	SourceExternalID    string
+	TargetExternalID    string
+	ConditionSensorType string
+	ConditionOperator   string
+	ConditionValue      float64
+	ActionType          string
+	ActionParams        []byte // raw JSON, decoded during index build
+}
+
+// Fetcher loads all currently active rules. The real implementation queries
+// PostgreSQL; tests supply a fake.
+type Fetcher interface {
+	FetchActiveRules(ctx context.Context) ([]Row, error)
+}
+
+type Cache struct {
+	fetcher  Fetcher
+	interval time.Duration
+	logger   *slog.Logger
+
+	mu    sync.RWMutex
+	index map[string][]rules.Rule
+
+	stop chan struct{}
+	done chan struct{}
+}
+
+func New(fetcher Fetcher, interval time.Duration, logger *slog.Logger) *Cache {
+	return &Cache{fetcher: fetcher, interval: interval, logger: logger}
+}
+
+// Start performs a synchronous initial load (so the service never runs with
+// an empty cache) and then begins the periodic refresh loop.
+func (c *Cache) Start(ctx context.Context) error {
+	if err := c.refresh(ctx); err != nil {
+		return err
+	}
+	c.stop = make(chan struct{})
+	c.done = make(chan struct{})
+	go c.loop()
+	return nil
+}
+
+func (c *Cache) Stop() {
+	close(c.stop)
+	<-c.done
+}
+
+func (c *Cache) loop() {
+	defer close(c.done)
+	ticker := time.NewTicker(c.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			ctx, cancel := context.WithTimeout(context.Background(), c.interval)
+			if err := c.refresh(ctx); err != nil {
+				c.logger.Error("rule cache refresh failed", "error", err)
+			}
+			cancel()
+		case <-c.stop:
+			return
+		}
+	}
+}
+
+func (c *Cache) refresh(ctx context.Context) error {
+	dbRows, err := c.fetcher.FetchActiveRules(ctx)
+	if err != nil {
+		return err
+	}
+	index := buildIndex(dbRows, c.logger)
+	c.mu.Lock()
+	c.index = index
+	c.mu.Unlock()
+	return nil
+}
+
+// MatchingRules returns the active rules that watch sensorType readings from
+// deviceID. Safe for concurrent use.
+func (c *Cache) MatchingRules(deviceID, sensorType string) []rules.Rule {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.index[cacheKey(deviceID, sensorType)]
+}
+
+func cacheKey(deviceID, sensorType string) string {
+	return deviceID + "\x00" + sensorType
+}
+
+func buildIndex(dbRows []Row, logger *slog.Logger) map[string][]rules.Rule {
+	index := make(map[string][]rules.Rule)
+	for _, r := range dbRows {
+		var params map[string]any
+		if len(r.ActionParams) > 0 {
+			if err := json.Unmarshal(r.ActionParams, &params); err != nil {
+				logger.Error("skipping rule with invalid action_params", "rule_id", r.RuleID, "error", err)
+				continue
+			}
+		}
+
+		rule := rules.Rule{
+			ID:                  r.RuleID,
+			ZoneID:              r.ZoneID,
+			SourceDeviceID:      r.SourceExternalID,
+			ConditionSensorType: r.ConditionSensorType,
+			ConditionOperator:   r.ConditionOperator,
+			ConditionValue:      r.ConditionValue,
+			TargetDeviceID:      r.TargetExternalID,
+			ActionType:          r.ActionType,
+			ActionParams:        params,
+		}
+
+		key := cacheKey(r.SourceExternalID, r.ConditionSensorType)
+		index[key] = append(index[key], rule)
+	}
+	return index
+}

services/rule-engine-service/internal/rulecache/cache_test.go

@@ -0,0 +1,110 @@
+package rulecache
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+)
+
+type fakeFetcher struct {
+	mu    sync.Mutex
+	rows  []Row
+	calls chan struct{}
+}
+
+func (f *fakeFetcher) setRows(rows []Row) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.rows = rows
+}
+
+func (f *fakeFetcher) FetchActiveRules(context.Context) ([]Row, error) {
+	f.mu.Lock()
+	rows := append([]Row(nil), f.rows...)
+	f.mu.Unlock()
+	if f.calls != nil {
+		f.calls <- struct{}{}
+	}
+	return rows, nil
+}
+
+func discardLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func TestCache_StartLoadsSynchronously(t *testing.T) {
+	fetcher := &fakeFetcher{rows: []Row{
+		{RuleID: 1, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", ConditionOperator: ">", ConditionValue: 28, TargetExternalID: "fan-1", ActionType: "turn_on"},
+	}}
+	c := New(fetcher, time.Hour, discardLogger())
+
+	if err := c.Start(context.Background()); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	defer c.Stop()
+
+	got := c.MatchingRules("sensor-1", "temperature")
+	if len(got) != 1 || got[0].TargetDeviceID != "fan-1" {
+		t.Fatalf("got %+v, want one rule targeting fan-1", got)
+	}
+
+	// Different sensor_type on the same device must not match.
+	if got := c.MatchingRules("sensor-1", "humidity"); len(got) != 0 {
+		t.Fatalf("got %+v, want no rules for humidity", got)
+	}
+}
+
+func TestCache_SkipsRuleWithInvalidActionParams(t *testing.T) {
+	fetcher := &fakeFetcher{rows: []Row{
+		{RuleID: 1, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", ActionParams: []byte(`not json`)},
+		{RuleID: 2, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", ActionParams: []byte(`{"level":50}`)},
+	}}
+	c := New(fetcher, time.Hour, discardLogger())
+	if err := c.Start(context.Background()); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	defer c.Stop()
+
+	got := c.MatchingRules("sensor-1", "temperature")
+	if len(got) != 1 || got[0].ID != 2 {
+		t.Fatalf("got %+v, want only rule 2 to survive", got)
+	}
+}
+
+func TestCache_RefreshesOnInterval(t *testing.T) {
+	calls := make(chan struct{}, 4)
+	fetcher := &fakeFetcher{calls: calls}
+	c := New(fetcher, 20*time.Millisecond, discardLogger())
+	if err := c.Start(context.Background()); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	defer c.Stop()
+
+	<-calls // consume the initial synchronous load
+
+	fetcher.setRows([]Row{
+		{RuleID: 1, SourceExternalID: "sensor-1", ConditionSensorType: "temperature", TargetExternalID: "fan-1", ActionType: "turn_on"},
+	})
+
+	select {
+	case <-calls:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for periodic refresh")
+	}
+
+	got := c.MatchingRules("sensor-1", "temperature")
+	if len(got) != 1 {
+		t.Fatalf("got %+v after refresh, want the newly added rule", got)
+	}
+}
+
+func TestCache_StopWaitsForLoopExit(t *testing.T) {
+	c := New(&fakeFetcher{}, time.Hour, discardLogger())
+	if err := c.Start(context.Background()); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	c.Stop() // must return without hanging
+}

services/rule-engine-service/internal/rulecache/postgres.go

@@ -0,0 +1,50 @@
+package rulecache
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// activeRulesQuery resolves condition_source_device_id/target_device_id (FKs
+// into devices.id) to devices.external_id, since every other part of the
+// platform (MQTT, Redis, gRPC) keys devices by their external ID, not
+// PostgreSQL's internal bigserial ID.
+const activeRulesQuery = `
+SELECT
+	r.id AS rule_id,
+	r.zone_id AS zone_id,
+	src.external_id AS source_external_id,
+	tgt.external_id AS target_external_id,
+	r.condition_sensor_type AS condition_sensor_type,
+	r.condition_operator AS condition_operator,
+	r.condition_value AS condition_value,
+	r.action_type AS action_type,
+	r.action_params AS action_params
+FROM automation_rules r
+JOIN devices src ON src.id = r.condition_source_device_id
+JOIN devices tgt ON tgt.id = r.target_device_id
+WHERE r.is_active = true
+`
+
+type PostgresFetcher struct {
+	pool *pgxpool.Pool
+}
+
+func NewPostgresFetcher(pool *pgxpool.Pool) *PostgresFetcher {
+	return &PostgresFetcher{pool: pool}
+}
+
+func (f *PostgresFetcher) FetchActiveRules(ctx context.Context) ([]Row, error) {
+	dbRows, err := f.pool.Query(ctx, activeRulesQuery)
+	if err != nil {
+		return nil, fmt.Errorf("query active rules: %w", err)
+	}
+	rows, err := pgx.CollectRows(dbRows, pgx.RowToStructByName[Row])
+	if err != nil {
+		return nil, fmt.Errorf("scan active rules: %w", err)
+	}
+	return rows, nil
+}

services/rule-engine-service/internal/rules/rule.go

@@ -0,0 +1,41 @@
+// Package rules holds the generic automation-rule model and condition
+// evaluation logic. Nothing here knows about growboxes, lights, or pumps —
+// only sensor_type/operator/value, exactly as stored in automation_rules.
+package rules
+
+import "fmt"
+
+// Rule mirrors one row of automation_rules, joined against devices so the
+// device IDs are the external identifiers used by MQTT/Redis/gRPC (not
+// PostgreSQL's internal bigserial IDs).
+type Rule struct {
+	ID                  int64
+	ZoneID              int64
+	SourceDeviceID      string // condition_source_device_id, resolved to devices.external_id
+	ConditionSensorType string
+	ConditionOperator   string
+	ConditionValue      float64
+	TargetDeviceID      string // target_device_id, resolved to devices.external_id
+	ActionType          string
+	ActionParams        map[string]any
+}
+
+// Evaluate reports whether value satisfies the rule's condition.
+func Evaluate(operator string, value, threshold float64) (bool, error) {
+	switch operator {
+	case ">":
+		return value > threshold, nil
+	case "<":
+		return value < threshold, nil
+	case ">=":
+		return value >= threshold, nil
+	case "<=":
+		return value <= threshold, nil
+	case "=":
+		return value == threshold, nil
+	case "!=":
+		return value != threshold, nil
+	default:
+		return false, fmt.Errorf("unknown condition operator %q", operator)
+	}
+}

services/rule-engine-service/internal/rules/rule_test.go

@@ -0,0 +1,39 @@
+package rules
+
+import "testing"
+
+func TestEvaluate(t *testing.T) {
+	cases := []struct {
+		operator  string
+		value     float64
+		threshold float64
+		want      bool
+	}{
+		{">", 29, 28, true},
+		{">", 27, 28, false},
+		{"<", 27, 28, true},
+		{"<", 29, 28, false},
+		{">=", 28, 28, true},
+		{"<=", 28, 28, true},
+		{"=", 28, 28, true},
+		{"=", 28.1, 28, false},
+		{"!=", 28.1, 28, true},
+		{"!=", 28, 28, false},
+	}
+
+	for _, c := range cases {
+		got, err := Evaluate(c.operator, c.value, c.threshold)
+		if err != nil {
+			t.Fatalf("operator %q: unexpected error: %v", c.operator, err)
+		}
+		if got != c.want {
+			t.Errorf("Evaluate(%q, %v, %v) = %v, want %v", c.operator, c.value, c.threshold, got, c.want)
+		}
+	}
+}
+
+func TestEvaluate_UnknownOperator(t *testing.T) {
+	if _, err := Evaluate("~=", 1, 2); err == nil {
+		t.Fatal("expected error for unknown operator")
+	}
+}