diff --git a/.env.example b/.env.example index b32d905..058b127 100644 --- a/.env.example +++ b/.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 diff --git a/docker-compose.yml b/docker-compose.yml index 02a1a03..7e67f7c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -147,6 +147,34 @@ services: - 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. + rule-engine-service: + build: + 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. diff --git a/services/rule-engine-service/Dockerfile b/services/rule-engine-service/Dockerfile new file mode 100644 index 0000000..5b67b1e --- /dev/null +++ b/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"] diff --git a/services/rule-engine-service/README.md b/services/rule-engine-service/README.md index 86d8ebc..e51d80b 100644 --- a/services/rule-engine-service/README.md +++ b/services/rule-engine-service/README.md @@ -1,18 +1,52 @@ # rule-engine-service (Go) -Статус: пока не реализован. +Статус: реализован (MVP). Зона ответственности: -- Потребляет события «новое показание» из RabbitMQ (асинхронно — здесь - надёжность доставки важнее задержки). -- Загружает активные `automation_rules` для нужной зоны/устройства из - PostgreSQL, кэшируя в памяти/Redis, чтобы не ходить в БД на каждое событие. +- Потребляет события «новое показание» из RabbitMQ (очередь + `telemetry.new_reading`, публикуемая ingest-service) — асинхронно, с ручным + ack/nack: надёжность доставки здесь важнее задержки. +- Загружает активные `automation_rules` из PostgreSQL в кэш в памяти + (обновляется по тикеру), чтобы не ходить в БД на каждое событие. - Оценивает условия правил обобщённо: `sensor_type X operator value` против показания — никаких зашитых названий датчиков/устройств, всё берётся из строки `automation_rules`. - При срабатывании: вызывает device-control-service по gRPC (нужен быстрый - результат успех/неудача) и публикует событие в notification-service через - RabbitMQ (не критично к задержке). + результат успех/неудача) и публикует событие `automation.rule_triggered` в + 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`. diff --git a/services/rule-engine-service/cmd/rule-engine-service/main.go b/services/rule-engine-service/cmd/rule-engine-service/main.go new file mode 100644 index 0000000..5989360 --- /dev/null +++ b/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 +} diff --git a/services/rule-engine-service/go.mod b/services/rule-engine-service/go.mod new file mode 100644 index 0000000..a12021e --- /dev/null +++ b/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 diff --git a/services/rule-engine-service/go.sum b/services/rule-engine-service/go.sum new file mode 100644 index 0000000..a5ecd8b --- /dev/null +++ b/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= diff --git a/services/rule-engine-service/internal/config/config.go b/services/rule-engine-service/internal/config/config.go new file mode 100644 index 0000000..72cd856 --- /dev/null +++ b/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 +} diff --git a/services/rule-engine-service/internal/devicecontrolclient/client.go b/services/rule-engine-service/internal/devicecontrolclient/client.go new file mode 100644 index 0000000..09ebbf3 --- /dev/null +++ b/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 + } +} diff --git a/services/rule-engine-service/internal/devicecontrolclient/client_test.go b/services/rule-engine-service/internal/devicecontrolclient/client_test.go new file mode 100644 index 0000000..55013b3 --- /dev/null +++ b/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) + } +} diff --git a/services/rule-engine-service/internal/engine/engine.go b/services/rule-engine-service/internal/engine/engine.go new file mode 100644 index 0000000..1c72acd --- /dev/null +++ b/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 +} diff --git a/services/rule-engine-service/internal/engine/engine_test.go b/services/rule-engine-service/internal/engine/engine_test.go new file mode 100644 index 0000000..c91b22b --- /dev/null +++ b/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) + } +} diff --git a/services/rule-engine-service/internal/rabbitmq/client.go b/services/rule-engine-service/internal/rabbitmq/client.go new file mode 100644 index 0000000..9b39f07 --- /dev/null +++ b/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() +} diff --git a/services/rule-engine-service/internal/rabbitmq/consumer.go b/services/rule-engine-service/internal/rabbitmq/consumer.go new file mode 100644 index 0000000..a643e3d --- /dev/null +++ b/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) + } + } + } +} diff --git a/services/rule-engine-service/internal/rabbitmq/publisher.go b/services/rule-engine-service/internal/rabbitmq/publisher.go new file mode 100644 index 0000000..f70d94e --- /dev/null +++ b/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 +} diff --git a/services/rule-engine-service/internal/rulecache/cache.go b/services/rule-engine-service/internal/rulecache/cache.go new file mode 100644 index 0000000..e084a7f --- /dev/null +++ b/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, ¶ms); 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 +} diff --git a/services/rule-engine-service/internal/rulecache/cache_test.go b/services/rule-engine-service/internal/rulecache/cache_test.go new file mode 100644 index 0000000..b1f6391 --- /dev/null +++ b/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 +} diff --git a/services/rule-engine-service/internal/rulecache/postgres.go b/services/rule-engine-service/internal/rulecache/postgres.go new file mode 100644 index 0000000..07a5da8 --- /dev/null +++ b/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 +} diff --git a/services/rule-engine-service/internal/rules/rule.go b/services/rule-engine-service/internal/rules/rule.go new file mode 100644 index 0000000..1f9ba14 --- /dev/null +++ b/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) + } +} diff --git a/services/rule-engine-service/internal/rules/rule_test.go b/services/rule-engine-service/internal/rules/rule_test.go new file mode 100644 index 0000000..234d5b4 --- /dev/null +++ b/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") + } +}