home_automatization/services/rule-engine-service/internal/engine/engine.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/metrics"
	"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 {
	metrics.ReadingsConsumed.Inc()

	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
		}
		metrics.RulesMatched.Inc()

		start := time.Now()
		result, dispatchErr := e.dispatcher.Dispatch(ctx, rule.TargetDeviceID, rule.ActionType, rule.ActionParams)
		metrics.DispatchDuration.WithLabelValues(rule.ActionType).Observe(time.Since(start).Seconds())

		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)
			metrics.RulesTriggered.WithLabelValues(rule.ActionType, "transport_error").Inc()
			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)
			metrics.RulesTriggered.WithLabelValues(rule.ActionType, "rejected").Inc()
		default:
			e.logger.Info("rule triggered", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "action_type", rule.ActionType)
			metrics.RulesTriggered.WithLabelValues(rule.ActionType, "success").Inc()
		}

		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
}