Primora/apps/backend/internal/services/alerting.go

package services

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"

	db "github.com/tdvorak/primora/apps/backend/internal/database/db"
	"github.com/tdvorak/primora/apps/backend/internal/models"
)

// Alert rules detect conditions in telemetry (component heartbeat silence,
// error-rate spikes) and fire `alert.fired` / `alert.resolved` domain events.
// Delivery reuses the webhook dispatcher + realtime stream — the evaluator is
// the only new machinery. Rules fire on state transitions only, so a flaky
// component produces one fired + one resolved, not a storm.

const (
	AlertEventFired    = "alert.fired"
	AlertEventResolved = "alert.resolved"

	AlertKindHeartbeatSilence = "heartbeat_silence"
	AlertKindErrorSpike       = "error_spike"
)

// AlertRuleConfig is the JSONB config per rule kind.
//
//	heartbeat_silence: component beats less often than ThresholdMinutes.
//	error_spike:       component logs more than ThresholdCount errors in
//	                   WindowMinutes.
// Component empty = evaluate every component in the project.
type AlertRuleConfig struct {
	Component        string `json:"component,omitempty"`
	ThresholdMinutes int    `json:"threshold_minutes,omitempty"`
	ThresholdCount   int64  `json:"threshold_count,omitempty"`
	WindowMinutes    int    `json:"window_minutes,omitempty"`
}

type AlertRuleSummary struct {
	ID        string          `json:"id"`
	ProjectID string          `json:"project_id"`
	Name      string          `json:"name"`
	Kind      string          `json:"kind"`
	Config    AlertRuleConfig `json:"config"`
	Enabled   bool            `json:"enabled"`
	Firing    []string        `json:"firing"`
	CreatedAt time.Time       `json:"created_at"`
	UpdatedAt time.Time       `json:"updated_at"`
}

type UpsertAlertRuleInput struct {
	Name    string          `json:"name" validate:"required"`
	Kind    string          `json:"kind" validate:"required"`
	Config  AlertRuleConfig `json:"config"`
	Enabled *bool           `json:"enabled"`
}

func (s *PlatformService) validateAlertRule(input UpsertAlertRuleInput) (AlertRuleConfig, error) {
	if !projectSecretNameRe.MatchString(input.Name) {
		return AlertRuleConfig{}, fmt.Errorf("invalid alert rule name %q — use [A-Za-z_][A-Za-z0-9_]*", input.Name)
	}
	cfg := input.Config
	switch input.Kind {
	case AlertKindHeartbeatSilence:
		if cfg.ThresholdMinutes < 1 || cfg.ThresholdMinutes > 10080 {
			return cfg, errors.New("threshold_minutes must be 1–10080")
		}
	case AlertKindErrorSpike:
		if cfg.ThresholdCount < 1 {
			return cfg, errors.New("threshold_count must be ≥ 1")
		}
		if cfg.WindowMinutes == 0 {
			cfg.WindowMinutes = 15
		}
		if cfg.WindowMinutes < 1 || cfg.WindowMinutes > 1440 {
			return cfg, errors.New("window_minutes must be 1–1440")
		}
	default:
		return cfg, fmt.Errorf("invalid kind %q — %s or %s", input.Kind, AlertKindHeartbeatSilence, AlertKindErrorSpike)
	}
	if cfg.Component != "" && len(cfg.Component) > 128 {
		return cfg, errors.New("component name too long")
	}
	return cfg, nil
}

func alertRuleSummary(row db.CoreAlertRule) AlertRuleSummary {
	var cfg AlertRuleConfig
	_ = json.Unmarshal(row.Config, &cfg)
	firing := parseFiring(row.State)
	return AlertRuleSummary{
		ID:        row.ID.String(),
		ProjectID: row.ProjectID.String(),
		Name:      row.Name,
		Kind:      row.Kind,
		Config:    cfg,
		Enabled:   row.Enabled,
		Firing:    firing.list(),
		CreatedAt: row.CreatedAt.Time,
		UpdatedAt: row.UpdatedAt.Time,
	}
}

func (s *PlatformService) ListAlertRules(ctx context.Context, actor *models.Actor, projectID uuid.UUID) ([]AlertRuleSummary, error) {
	if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
		return nil, err
	}
	rows, err := s.repo.Queries().ListAlertRules(ctx, projectID)
	if err != nil {
		return nil, err
	}
	out := make([]AlertRuleSummary, len(rows))
	for i, r := range rows {
		out[i] = alertRuleSummary(r)
	}
	return out, nil
}

func (s *PlatformService) CreateAlertRule(ctx context.Context, actor *models.Actor, projectID uuid.UUID, input UpsertAlertRuleInput, requestID string) (AlertRuleSummary, error) {
	if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
		return AlertRuleSummary{}, err
	}
	cfg, err := s.validateAlertRule(input)
	if err != nil {
		return AlertRuleSummary{}, err
	}
	rawCfg, _ := json.Marshal(cfg)
	row, err := s.repo.Queries().CreateAlertRule(ctx, db.CreateAlertRuleParams{
		ProjectID: projectID,
		Name:      input.Name,
		Kind:      input.Kind,
		Config:    rawCfg,
	})
	if err != nil {
		if strings.Contains(err.Error(), "alert_rules_project_id_name_key") {
			return AlertRuleSummary{}, fmt.Errorf("an alert rule named %s already exists", input.Name)
		}
		return AlertRuleSummary{}, err
	}
	s.auditAlert(ctx, actor, projectID, requestID, "alert.created", row.ID, input.Name)
	return alertRuleSummary(row), nil
}

func (s *PlatformService) UpdateAlertRule(ctx context.Context, actor *models.Actor, projectID, ruleID uuid.UUID, input UpsertAlertRuleInput, requestID string) (AlertRuleSummary, error) {
	if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
		return AlertRuleSummary{}, err
	}
	if _, err := s.repo.Queries().GetAlertRule(ctx, db.GetAlertRuleParams{ID: ruleID, ProjectID: projectID}); err != nil {
		return AlertRuleSummary{}, err // pgx.ErrNoRows → 404 at the handler
	}
	cfg, err := s.validateAlertRule(input)
	if err != nil {
		return AlertRuleSummary{}, err
	}
	enabled := true
	if input.Enabled != nil {
		enabled = *input.Enabled
	}
	rawCfg, _ := json.Marshal(cfg)
	row, err := s.repo.Queries().UpdateAlertRule(ctx, db.UpdateAlertRuleParams{
		ID: ruleID, ProjectID: projectID, Name: input.Name, Config: rawCfg, Enabled: enabled,
	})
	if err != nil {
		return AlertRuleSummary{}, err
	}
	s.auditAlert(ctx, actor, projectID, requestID, "alert.updated", row.ID, input.Name)
	return alertRuleSummary(row), nil
}

func (s *PlatformService) DeleteAlertRule(ctx context.Context, actor *models.Actor, projectID, ruleID uuid.UUID, requestID string) error {
	if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
		return err
	}
	if _, err := s.repo.Queries().GetAlertRule(ctx, db.GetAlertRuleParams{ID: ruleID, ProjectID: projectID}); err != nil {
		return err
	}
	if err := s.repo.Queries().DeleteAlertRule(ctx, db.DeleteAlertRuleParams{ID: ruleID, ProjectID: projectID}); err != nil {
		return err
	}
	s.auditAlert(ctx, actor, projectID, requestID, "alert.deleted", ruleID, "")
	return nil
}

func (s *PlatformService) auditAlert(ctx context.Context, actor *models.Actor, projectID uuid.UUID, requestID, action string, ruleID uuid.UUID, name string) {
	project, err := s.repo.Queries().GetProjectByID(ctx, projectID)
	if err != nil {
		return
	}
	_, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, action, "alert_rule", ruleID.String(), map[string]any{"name": name}))
}

// ---------------------------------------------------------------------------
// evaluator — one ticker per backend, transition-based firing
// ---------------------------------------------------------------------------

type firingSet map[string]bool

func parseFiring(raw []byte) firingSet {
	var state struct {
		Firing map[string]bool `json:"firing"`
	}
	if err := json.Unmarshal(raw, &state); err != nil || state.Firing == nil {
		return firingSet{}
	}
	return state.Firing
}

func (f firingSet) list() []string {
	out := make([]string, 0, len(f))
	for name, on := range f {
		if on {
			out = append(out, name)
		}
	}
	return out
}

// alertQueries is the slice of db.Queries the evaluator needs — narrow so
// tests can stub it without a database.
type alertQueries interface {
	ListEnabledAlertRules(ctx context.Context) ([]db.CoreAlertRule, error)
	LastHeartbeats(ctx context.Context, projectID uuid.UUID) ([]db.LastHeartbeatsRow, error)
	RecentErrorCounts(ctx context.Context, arg db.RecentErrorCountsParams) ([]db.RecentErrorCountsRow, error)
	SetAlertRuleState(ctx context.Context, arg db.SetAlertRuleStateParams) error
}

type alertEvaluator struct {
	q       alertQueries
	publish func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any)
	logger  *slog.Logger
	tick    time.Duration
	quit    chan struct{}
	wg      sync.WaitGroup
	started sync.Once
	leader  *leaderState
}

func newAlertEvaluator(q alertQueries, publish func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any), logger *slog.Logger, pool *pgxpool.Pool) *alertEvaluator {
	return &alertEvaluator{q: q, publish: publish, logger: logger, tick: time.Minute, quit: make(chan struct{}), leader: newLeaderState(pool, leaderKeyAlerts)}
}

func (e *alertEvaluator) Start(ctx context.Context) {
	e.started.Do(func() {
		e.wg.Add(1)
		go func() {
			defer e.wg.Done()
			t := time.NewTicker(e.tick)
			defer t.Stop()
			for {
				select {
				case <-t.C:
					if e.leader.hold(ctx) {
						e.scan(ctx)
					}
				case <-e.quit:
					e.leader.stop(context.Background())
					return
				case <-ctx.Done():
					return
				}
			}
		}()
	})
}

func (e *alertEvaluator) Stop() {
	close(e.quit)
	e.wg.Wait()
}

func (e *alertEvaluator) scan(ctx context.Context) {
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	rules, err := e.q.ListEnabledAlertRules(ctx)
	if err != nil {
		e.logger.Warn("alert scan: list rules", "error", err)
		return
	}
	for _, rule := range rules {
		if err := e.evaluate(ctx, rule); err != nil {
			e.logger.Warn("alert scan: evaluate", "rule", rule.Name, "error", err)
		}
	}
}

func (e *alertEvaluator) evaluate(ctx context.Context, rule db.CoreAlertRule) error {
	var cfg AlertRuleConfig
	if err := json.Unmarshal(rule.Config, &cfg); err != nil {
		return nil // bad config can't be fixed by re-scanning; skip quietly
	}
	now := time.Now()
	var breached map[string]float64 // component → measured value
	var metric, unit string

	switch rule.Kind {
	case AlertKindHeartbeatSilence:
		rows, err := e.q.LastHeartbeats(ctx, rule.ProjectID)
		if err != nil {
			return err
		}
		breached = map[string]float64{}
		for _, r := range rows {
			silent := now.Sub(r.LastBeat.Time).Minutes()
			if silent >= float64(cfg.ThresholdMinutes) {
				breached[r.ComponentName] = silent
			}
		}
		metric, unit = "silence", "minutes"

	case AlertKindErrorSpike:
		rows, err := e.q.RecentErrorCounts(ctx, db.RecentErrorCountsParams{
			ProjectID: rule.ProjectID,
			Column2:   int64(cfg.WindowMinutes) * 60,
		})
		if err != nil {
			return err
		}
		breached = map[string]float64{}
		for _, r := range rows {
			if r.Errors >= cfg.ThresholdCount {
				breached[r.ComponentName] = float64(r.Errors)
			}
		}
		metric, unit = "errors", "count"

	default:
		return nil
	}

	firing := parseFiring(rule.State)
	changed := false
	// newly breached components → fired
	for comp, value := range breached {
		if cfg.Component != "" && comp != cfg.Component {
			continue
		}
		if firing[comp] {
			continue
		}
		firing[comp] = true
		changed = true
		e.publish(ctx, rule.ProjectID, AlertEventFired, map[string]any{
			"rule_id": rule.ID.String(), "rule": rule.Name, "kind": rule.Kind,
			"component": comp, "metric": metric, "value": value, "unit": unit,
		})
	}
	// recovered components → resolved
	for comp, on := range firing {
		_, stillBreached := breached[comp]
		if !on || stillBreached {
			continue
		}
		firing[comp] = false
		changed = true
		e.publish(ctx, rule.ProjectID, AlertEventResolved, map[string]any{
			"rule_id": rule.ID.String(), "rule": rule.Name, "kind": rule.Kind,
			"component": comp,
		})
	}
	if !changed {
		return nil
	}
	raw, _ := json.Marshal(map[string]any{"firing": firing})
	return e.q.SetAlertRuleState(ctx, db.SetAlertRuleStateParams{ID: rule.ID, ProjectID: rule.ProjectID, State: raw})
}