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. Показание ниже порога проверено отдельно — правило корректно не срабатывает.
49 lines
1.3 KiB
Go
49 lines
1.3 KiB
Go
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
|
|
}
|