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

// 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/metrics"
	"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 {
		metrics.CacheRefreshes.WithLabelValues("error").Inc()
		return err
	}
	index := buildIndex(dbRows, c.logger)
	c.mu.Lock()
	c.index = index
	c.mu.Unlock()
	metrics.CacheRefreshes.WithLabelValues("success").Inc()
	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
}