cacto
abc83faf05
Слушает 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. Показание ниже порога проверено отдельно — правило корректно не срабатывает.
97 lines
3.7 KiB
Go
97 lines
3.7 KiB
Go
// 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
|
|
}
|