// Package rulestore is pgx-backed CRUD for alert_rules/alert_state, plus // the two operations that make the evaluator's concurrency and delivery // guarantees hold (see /docs/phase-3-alerting-design.md's fixes 1 and 2): // ClaimDueRules (atomic claim-before-evaluate) and ApplyTransition (the // state transition and the delivery_log outbox insert in one DB // transaction). package rulestore import ( "context" "errors" "fmt" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) var ErrNotFound = errors.New("not found") type ConditionType string const ( ConditionThreshold ConditionType = "threshold" ConditionAbsence ConditionType = "absence" ) func ValidConditionType(c ConditionType) bool { return c == ConditionThreshold || c == ConditionAbsence } type Comparator string const ( Gt Comparator = "gt" Gte Comparator = "gte" Lt Comparator = "lt" Lte Comparator = "lte" Eq Comparator = "eq" Ne Comparator = "ne" ) func ValidComparator(c Comparator) bool { switch c { case Gt, Gte, Lt, Lte, Eq, Ne: return true default: return false } } type State string const ( StateOK State = "ok" StatePending State = "pending" StateFiring State = "firing" ) type Rule struct { ID string `json:"id"` TenantID string `json:"tenant_id"` Name string `json:"name"` Description string `json:"description"` Query string `json:"query"` QueryLanguage string `json:"query_language"` ConditionType ConditionType `json:"condition_type"` Comparator *Comparator `json:"comparator,omitempty"` ThresholdValue *float64 `json:"threshold_value,omitempty"` EvalIntervalSeconds int `json:"eval_interval_seconds"` ForMinutes int `json:"for_minutes"` RenotifyIntervalMinutes *int `json:"renotify_interval_minutes,omitempty"` NotificationTargetID string `json:"notification_target_id"` Enabled bool `json:"enabled"` CreatedBy string `json:"created_by"` } type AlertState struct { RuleID string `json:"rule_id"` State State `json:"state"` ConditionTrueSince *time.Time `json:"condition_true_since,omitempty"` FiredAt *time.Time `json:"fired_at,omitempty"` LastNotifiedAt *time.Time `json:"last_notified_at,omitempty"` LastEvaluatedAt *time.Time `json:"last_evaluated_at,omitempty"` LastEvalStatus string `json:"last_eval_status"` LastError *string `json:"last_error,omitempty"` LastValue *float64 `json:"last_value,omitempty"` ConsecutiveErrors int `json:"consecutive_errors"` } // RuleWithState is a rule joined with its current live state, exactly // what the evaluator needs to run one evaluation and what the read API // returns. type RuleWithState struct { Rule State AlertState `json:"state"` } type Store struct { pool *pgxpool.Pool } func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} } // Create inserts the rule and its initial alert_state row in one // transaction. A rule with no alert_state row is silently never picked // up by ClaimDueRules -- see the design doc's explicit warning about // this exact failure mode. func (s *Store) Create(ctx context.Context, r *Rule) error { r.ID = uuid.NewString() if r.TenantID == "" { r.TenantID = "default" } if r.CreatedBy == "" { r.CreatedBy = "anonymous" } tx, err := s.pool.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) _, err = tx.Exec(ctx, ` INSERT INTO alert_rules (id, tenant_id, name, description, query, query_language, condition_type, comparator, threshold_value, eval_interval_seconds, for_minutes, renotify_interval_minutes, notification_target_id, enabled, created_by) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)`, r.ID, r.TenantID, r.Name, r.Description, r.Query, r.QueryLanguage, r.ConditionType, r.Comparator, r.ThresholdValue, r.EvalIntervalSeconds, r.ForMinutes, r.RenotifyIntervalMinutes, r.NotificationTargetID, r.Enabled, r.CreatedBy) if err != nil { return fmt.Errorf("inserting rule: %w", err) } _, err = tx.Exec(ctx, ` INSERT INTO alert_state (rule_id, tenant_id, state, last_eval_status, next_eval_at) VALUES ($1, $2, 'ok', 'ok', now())`, r.ID, r.TenantID) if err != nil { return fmt.Errorf("inserting initial alert_state: %w", err) } return tx.Commit(ctx) } func (s *Store) List(ctx context.Context) ([]RuleWithState, error) { rows, err := s.pool.Query(ctx, ruleWithStateSelect+" ORDER BY r.created_at DESC") if err != nil { return nil, err } defer rows.Close() var out []RuleWithState for rows.Next() { rs, err := scanRuleWithState(rows) if err != nil { return nil, err } out = append(out, rs) } return out, rows.Err() } func (s *Store) Get(ctx context.Context, id string) (*RuleWithState, error) { rows, err := s.pool.Query(ctx, ruleWithStateSelect+" WHERE r.id = $1", id) if err != nil { return nil, err } defer rows.Close() if !rows.Next() { return nil, ErrNotFound } rs, err := scanRuleWithState(rows) if err != nil { return nil, err } return &rs, nil } func (s *Store) Delete(ctx context.Context, id string) error { tag, err := s.pool.Exec(ctx, `DELETE FROM alert_rules WHERE id = $1`, id) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } const ruleWithStateSelect = ` SELECT r.id, r.tenant_id, r.name, r.description, r.query, r.query_language, r.condition_type, r.comparator, r.threshold_value, r.eval_interval_seconds, r.for_minutes, r.renotify_interval_minutes, r.notification_target_id, r.enabled, r.created_by, s.state, s.condition_true_since, s.fired_at, s.last_notified_at, s.last_evaluated_at, s.last_eval_status, s.last_error, s.last_value, s.consecutive_errors FROM alert_rules r JOIN alert_state s ON s.rule_id = r.id` func scanRuleWithState(rows pgx.Rows) (RuleWithState, error) { var rs RuleWithState err := rows.Scan( &rs.ID, &rs.TenantID, &rs.Name, &rs.Description, &rs.Query, &rs.QueryLanguage, &rs.ConditionType, &rs.Comparator, &rs.ThresholdValue, &rs.EvalIntervalSeconds, &rs.ForMinutes, &rs.RenotifyIntervalMinutes, &rs.NotificationTargetID, &rs.Enabled, &rs.CreatedBy, &rs.State.State, &rs.State.ConditionTrueSince, &rs.State.FiredAt, &rs.State.LastNotifiedAt, &rs.State.LastEvaluatedAt, &rs.State.LastEvalStatus, &rs.State.LastError, &rs.State.LastValue, &rs.State.ConsecutiveErrors) rs.State.RuleID = rs.ID return rs, err } // ClaimDueRules atomically claims up to batchSize due, enabled rules -- // see /docs/phase-3-alerting-design.md's fix 1. next_eval_at is bumped // forward *before* returning, so a second scheduler tick (or a second // evaluator replica, later) can't re-select the same rule while this // one is still being evaluated. SKIP LOCKED means a concurrent claim // query never blocks on rows another claim already has locked -- it // just moves on to the next candidate. func (s *Store) ClaimDueRules(ctx context.Context, batchSize int) ([]RuleWithState, error) { rows, err := s.pool.Query(ctx, ` WITH due AS ( SELECT s.rule_id FROM alert_state s JOIN alert_rules r ON r.id = s.rule_id WHERE s.next_eval_at <= now() AND r.enabled ORDER BY s.next_eval_at LIMIT $1 FOR UPDATE OF s SKIP LOCKED ) UPDATE alert_state s SET next_eval_at = now() + make_interval(secs => r.eval_interval_seconds), claimed_at = now() FROM due, alert_rules r WHERE s.rule_id = due.rule_id AND r.id = s.rule_id RETURNING r.id, r.tenant_id, r.name, r.description, r.query, r.query_language, r.condition_type, r.comparator, r.threshold_value, r.eval_interval_seconds, r.for_minutes, r.renotify_interval_minutes, r.notification_target_id, r.enabled, r.created_by, s.state, s.condition_true_since, s.fired_at, s.last_notified_at, s.last_evaluated_at, s.last_eval_status, s.last_error, s.last_value, s.consecutive_errors`, batchSize) if err != nil { return nil, err } defer rows.Close() var out []RuleWithState for rows.Next() { rs, err := scanRuleWithState(rows) if err != nil { return nil, err } out = append(out, rs) } return out, rows.Err() } // NotifyEvent, if non-nil, is inserted into delivery_log in the same // transaction as the state update -- the transactional outbox from fix // 2. Payload is the already-rendered notification body (rulestore // doesn't know how to format one; that's internal/delivery's job). type NotifyEvent struct { NotificationTargetID string EventType string // "firing" | "resolved" Payload []byte // JSON } // ApplyTransition writes a successful evaluation's outcome: the new // alert_state fields, and -- atomically, in the same transaction -- a // pending delivery_log row if notify is non-nil. See fix 2: "decided to // notify" becomes durable exactly once, in lockstep with the state // change, before any network call to a notification target is // attempted. func (s *Store) ApplyTransition(ctx context.Context, ruleID string, next AlertState, notify *NotifyEvent) error { tx, err := s.pool.Begin(ctx) if err != nil { return err } defer tx.Rollback(ctx) _, err = tx.Exec(ctx, ` UPDATE alert_state SET state = $1, condition_true_since = $2, fired_at = $3, last_notified_at = $4, last_evaluated_at = now(), last_eval_status = 'ok', last_error = NULL, last_value = $5, consecutive_errors = 0 WHERE rule_id = $6`, next.State, next.ConditionTrueSince, next.FiredAt, next.LastNotifiedAt, next.LastValue, ruleID) if err != nil { return fmt.Errorf("updating alert_state: %w", err) } if notify != nil { _, err = tx.Exec(ctx, ` INSERT INTO delivery_log (rule_id, tenant_id, notification_target_id, event_type, status, next_attempt_at, payload) VALUES ($1, (SELECT tenant_id FROM alert_rules WHERE id = $1), $2, $3, 'pending', now(), $4)`, ruleID, notify.NotificationTargetID, notify.EventType, notify.Payload) if err != nil { return fmt.Errorf("inserting delivery_log outbox row: %w", err) } } return tx.Commit(ctx) } // DeliveryLogEntry is the read shape the web UI's delivery log view and // the httpapi read endpoint use -- deliberately excludes `payload` // (the rendered notification body isn't needed for the list view, and // keeping it out of the default read path avoids echoing // target-specific formatting, e.g. a rendered webhook secret, back over // HTTP by default). type DeliveryLogEntry struct { ID int64 `json:"id"` RuleID string `json:"rule_id"` NotificationTargetID string `json:"notification_target_id"` EventType string `json:"event_type"` Status string `json:"status"` AttemptCount int `json:"attempt_count"` LastError *string `json:"last_error,omitempty"` ResponseStatus *int `json:"response_status,omitempty"` CreatedAt time.Time `json:"created_at"` } // ListForRule reads delivery_log's UI-visible role -- "why didn't I get // paged" -- most recent first. func (s *Store) ListForRule(ctx context.Context, ruleID string, limit int) ([]DeliveryLogEntry, error) { rows, err := s.pool.Query(ctx, ` SELECT id, rule_id, notification_target_id, event_type, status, attempt_count, last_error, response_status, created_at FROM delivery_log WHERE rule_id = $1 ORDER BY created_at DESC LIMIT $2`, ruleID, limit) if err != nil { return nil, err } defer rows.Close() var out []DeliveryLogEntry for rows.Next() { var e DeliveryLogEntry if err := rows.Scan(&e.ID, &e.RuleID, &e.NotificationTargetID, &e.EventType, &e.Status, &e.AttemptCount, &e.LastError, &e.ResponseStatus, &e.CreatedAt); err != nil { return nil, err } out = append(out, e) } return out, rows.Err() } // RecordError updates only the error-tracking fields on alert_state -- // state itself is left untouched, per fix 3: an evaluation error must // never transition state (an error is not "condition false"). func (s *Store) RecordError(ctx context.Context, ruleID, errMsg string) error { _, err := s.pool.Exec(ctx, ` UPDATE alert_state SET last_evaluated_at = now(), last_eval_status = 'error', last_error = $1, consecutive_errors = consecutive_errors + 1 WHERE rule_id = $2`, errMsg, ruleID) return err }