refactor[plugins](alerts): group caches and tag-rule engine into cache.go and tagrules.go

Author: Kbayero

backend/modules/alerts/connectors/repository.go

@@ -17,12 +17,6 @@ type AlertRepository interface {
 	CountOpenAlerts(ctx context.Context) (int64, error)
 	CountByStatus(ctx context.Context, status int) (int64, error)
 	SearchByIDs(ctx context.Context, alertIDs []string) ([]domain.UtmAlert, error)
-	AssignAssetGroups(ctx context.Context, mapping map[string]AssetGroupRef) error
-}
-
-type AssetGroupRef struct {
-	ID   int64
-	Name string
 }
 
 type HistoryEntry struct {

backend/modules/alerts/module.go

@@ -1,9 +1,6 @@
 package alerts
 
 import (
-	"context"
-
-	"github.com/threatwinds/go-sdk/catcher"
 	"github.com/utmstack/utmstack/backend/modules/alerts/connectors"
 	"github.com/utmstack/utmstack/backend/modules/alerts/handler"
 	"github.com/utmstack/utmstack/backend/modules/alerts/repository"
@@ -17,14 +14,9 @@ type Module struct {
 	alertTagHandler     *handler.AlertTagHandler
 	alertTagRuleHandler *handler.AlertTagRuleHandler
 	adversaryHandler    *handler.AdversaryHandler
-	scheduler           *usecase.Scheduler
-	schedulerEnabled    bool
 }
 
-func NewModule(
-	db *gorm.DB,
-	schedulerEnabled bool,
-) *Module {
+func NewModule(db *gorm.DB) *Module {
 	alertRepo := repository.NewOSAlertRepository()
 
 	historyRecorder := repository.NewHistoryRecorder()
@@ -40,8 +32,6 @@ func NewModule(
 	alertTagRuleUC := usecase.NewAlertTagRuleUsecase(alertTagRuleRepo, alertTagRepo)
 	alertTagRuleH := handler.NewAlertTagRuleHandler(alertTagRuleUC)
 
-	sched := usecase.NewScheduler(alertRepo)
-
 	adversaryUC := usecase.NewAdversaryUsecase()
 	adversaryH := handler.NewAdversaryHandler(adversaryUC)
 
@@ -51,18 +41,7 @@ func NewModule(
 		alertTagHandler:     alertTagH,
 		alertTagRuleHandler: alertTagRuleH,
 		adversaryHandler:    adversaryH,
-		scheduler:           sched,
-		schedulerEnabled:    schedulerEnabled,
-	}
-}
-
-func (m *Module) Start(ctx context.Context) {
-	if !m.schedulerEnabled {
-		catcher.Info("alerts scheduler: disabled (ALERTS_SCHEDULER_ENABLED=false)", nil)
-		return
 	}
-	catcher.Info("alerts scheduler: enabled — launching goroutine", nil)
-	go m.scheduler.Start(ctx)
 }
 
 func (m *Module) GetAlertHandler() *handler.AlertHandler { return m.alertHandler }
@@ -76,5 +55,3 @@ func (m *Module) GetAlertTagRuleHandler() *handler.AlertTagRuleHandler {
 }
 
 func (m *Module) GetAdversaryHandler() *handler.AdversaryHandler { return m.adversaryHandler }
-
-func (m *Module) IsSchedulerEnabled() bool { return m.schedulerEnabled }

backend/modules/alerts/repository/alert_os.go

@@ -48,13 +48,6 @@ ctx._source.incidentDetail.createdBy    = params.createdBy;
 ctx._source.incidentDetail.source       = params.source;
 `
 
-const assignAssetGroupsScript = `
-if (params.mapping.containsKey(ctx._source.dataSource)) {
-  ctx._source.assetGroupId = params.mapping[ctx._source.dataSource].id;
-  ctx._source.assetGroupName = params.mapping[ctx._source.dataSource].name;
-}
-`
-
 // ---------------------------------------------------------------------------
 // osAlertRepo implements connectors.AlertRepository against the go-sdk `os` client.
 // ---------------------------------------------------------------------------
@@ -195,20 +188,3 @@ func (r *osAlertRepo) SearchByIDs(ctx context.Context, alertIDs []string) ([]dom
 	return alerts, nil
 }
 
-func (r *osAlertRepo) AssignAssetGroups(ctx context.Context, mapping map[string]connectors.AssetGroupRef) error {
-	filter := termQuery("status", int(domain.AlertStatusAutomaticReview))
-
-	// Convert mapping to a plain map[string]any for Painless params.
-	pmapping := make(map[string]any, len(mapping))
-	for k, v := range mapping {
-		pmapping[k] = map[string]any{"id": v.ID, "name": v.Name}
-	}
-
-	script := Script{
-		Source: assignAssetGroupsScript,
-		Params: map[string]any{
-			"mapping": pmapping,
-		},
-	}
-	return osUpdateByQuery(ctx, alertIndex, filter, script)
-}

backend/modules/alerts/repository/osquery.go

@@ -25,10 +25,6 @@ func termsQuery(field string, values []string) map[string]any {
 	return map[string]any{"terms": map[string]any{field: iValues}}
 }
 
-func matchQuery(field string, value any) map[string]any {
-	return map[string]any{"match": map[string]any{field: value}}
-}
-
 // Script holds a Painless script source and its parameters. Always populate
 // Params — never embed user input directly in Source.
 type Script struct {

backend/modules/alerts/usecase/scheduler.go

@@ -18,19 +18,16 @@ const (
 	defaultRefreshSec   = 60
 	rulesRequestTimeout = 10 * time.Second
 	activeRulesPath     = "/api/v1/internal/alert-tag-rules/active"
+	enrichmentPath      = "/api/v1/datasources/enrichment"
 )
 
-// RuleSnapshot is the per-rule view the plugin needs at evaluation time.
-// Conditions is left as raw JSON until used so we don't decode work we won't
-// run (a rule with empty conditions never reaches the evaluator).
 type RuleSnapshot struct {
 	ID         uint64
 	Name       string
 	Conditions []FilterType
 	TagNames   []string
 }
 
-// activeRuleWire matches the backend dto.ActiveAlertTagRule payload.
 type activeRuleWire struct {
 	ID              uint64          `json:"id"`
 	Name            string          `json:"name"`
@@ -161,3 +158,128 @@ func (c *ruleCache) Run(ctx context.Context) {
 		}
 	}
 }
+
+type datasourceEnrichment struct {
+	GroupID   *uint64
+	GroupName string
+	Labels    []string
+}
+
+type enrichmentWire struct {
+	Name      string   `json:"name"`
+	DataType  string   `json:"dataType"`
+	GroupID   *uint64  `json:"groupId"`
+	GroupName string   `json:"groupName"`
+	Labels    []string `json:"labels"`
+}
+
+type datasourceCache struct {
+	baseURL     string
+	internalKey string
+	httpClient  *http.Client
+	refresh     time.Duration
+
+	mu     sync.RWMutex
+	byName map[string]datasourceEnrichment
+}
+
+func newDatasourceCache() *datasourceCache {
+	cfg := plugins.PluginCfg("com.utmstack")
+	base := cfg.Get("backend").String()
+	if base != "" && !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") {
+		base = "http://" + base
+	}
+	refresh := time.Duration(defaultRefreshSec) * time.Second
+	if v := cfg.Get("rulesRefreshSec").Int(); v > 0 {
+		refresh = time.Duration(v) * time.Second
+	}
+	return &datasourceCache{
+		baseURL:     base,
+		internalKey: cfg.Get("internalKey").String(),
+		httpClient:  &http.Client{Timeout: rulesRequestTimeout},
+		refresh:     refresh,
+	}
+}
+
+func (c *datasourceCache) Lookup(name string) (datasourceEnrichment, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	e, ok := c.byName[name]
+	return e, ok
+}
+
+func (c *datasourceCache) Refresh(ctx context.Context) error {
+	if c.baseURL == "" || c.internalKey == "" {
+		return catcher.Error("datasource cache: backend URL or internal key missing", nil, map[string]any{"process": "plugin_com.utmstack.alerts"})
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+enrichmentPath, nil)
+	if err != nil {
+		return catcher.Error("datasource cache: build request failed", err, map[string]any{"process": "plugin_com.utmstack.alerts"})
+	}
+	req.Header.Set(internalKeyHeader, c.internalKey)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return catcher.Error("datasource cache: request failed", err, map[string]any{"process": "plugin_com.utmstack.alerts"})
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return catcher.Error("datasource cache: read body failed", err, map[string]any{"process": "plugin_com.utmstack.alerts"})
+	}
+	if resp.StatusCode >= 400 {
+		return catcher.Error("datasource cache: backend returned error", nil, map[string]any{
+			"status":  resp.StatusCode,
+			"body":    string(body),
+			"process": "plugin_com.utmstack.alerts",
+		})
+	}
+
+	var wire []enrichmentWire
+	if err := json.Unmarshal(body, &wire); err != nil {
+		return catcher.Error("datasource cache: decode failed", err, map[string]any{"process": "plugin_com.utmstack.alerts"})
+	}
+
+	next := make(map[string]datasourceEnrichment, len(wire))
+	for _, w := range wire {
+		// Keyed by name; a host's group is the same across its data types, so a
+		// repeated name just overwrites with equivalent enrichment.
+		next[w.Name] = datasourceEnrichment{
+			GroupID:   w.GroupID,
+			GroupName: w.GroupName,
+			Labels:    w.Labels,
+		}
+	}
+
+	c.mu.Lock()
+	c.byName = next
+	c.mu.Unlock()
+	return nil
+}
+
+func (c *datasourceCache) Run(ctx context.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			_ = catcher.Error("datasource cache: recovered from panic in Run", nil, map[string]any{
+				"panic":   r,
+				"process": "plugin_com.utmstack.alerts",
+			})
+		}
+	}()
+
+	ticker := time.NewTicker(c.refresh)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			refreshCtx, cancel := context.WithTimeout(context.Background(), rulesRequestTimeout)
+			_ = c.Refresh(refreshCtx)
+			cancel()
+		}
+	}
+}