Netpulse_SasS/server/internal/store/alerts_state.go

package store

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// Alert — рядок alr.alerts у вигляді, придатному і для доставки, і для UI.
type Alert struct {
	ID           string          `json:"id"`
	RuleID       string          `json:"rule_id,omitempty"`
	RuleName     string          `json:"rule_name,omitempty"`
	DeviceID     string          `json:"device_id,omitempty"`
	DeviceName   string          `json:"device_name,omitempty"`
	InterfaceID  string          `json:"interface_id,omitempty"`
	Severity     string          `json:"severity"`
	State        string          `json:"state"`
	Title        string          `json:"title"`
	Message      string          `json:"message,omitempty"`
	DedupKey     string          `json:"dedup_key"`
	Value        *float64        `json:"value,omitempty"`
	Threshold    *float64        `json:"threshold,omitempty"`
	Context      json.RawMessage `json:"context,omitempty"`
	RootAlertID  string          `json:"root_alert_id,omitempty"`
	SuppressedBy string          `json:"suppressed_by,omitempty"`
	StartedAt    time.Time       `json:"started_at"`
	AckedAt      *time.Time      `json:"acked_at,omitempty"`
	AckedByEmail string          `json:"acked_by_email,omitempty"`
	AckComment   string          `json:"ack_comment,omitempty"`
	ResolvedAt   *time.Time      `json:"resolved_at,omitempty"`
	LastSeenAt   time.Time       `json:"last_seen_at"`
	NotifyCount  int             `json:"notify_count"`
	// Скільки подій злилося в цей алерт. Для метричних завжди 1:
	// вимір не «стається», він просто триває.
	EventCount int `json:"event_count,omitempty"`
	// Стан драбини ескалації; порожньо — ескалації в цього алерту немає.
	// Без цього поля людина, підняту о третій ночі, не має де побачити,
	// звідки взявся дзвінок і коли буде наступний.
	Escalation *AlertEscalation `json:"escalation,omitempty"`
	// Тенант потрібен лише там, де алерти читаються наскрізь по всіх
	// кабінетах (черга розсилки подієвих) — у решті місць він відомий
	// із контексту запиту.
	TenantID string `json:"-"`
}

// FiredAlert — результат підняття.
type FiredAlert struct {
	Alert
	IsNew bool
	// PrevState — стан до цього тіку; порожній, якщо алерт щойно
	// створено. Потрібен, щоб відрізнити «нічого не змінилось» від
	// переходу firing↔suppressed: другий треба показати в UI, перший —
	// ні.
	PrevState string
}

// StateChanged — чи змінився стан наявного алерту.
func (f FiredAlert) StateChanged() bool {
	return !f.IsNew && f.PrevState != "" && f.PrevState != f.State
}

// RaiseAlert створює або продовжує алерт.
//
// Дедуплікація робиться індексом у БД, а не перевіркою в коді: два
// движки, що з якоїсь причини працюють одночасно, інакше створили б два
// алерти на ту саму проблему й розбудили б людину двічі.
func (s *Store) RaiseAlert(ctx context.Context, tenantID string, a Alert) (FiredAlert, error) {
	var out FiredAlert
	ctxJSON := "{}"
	if len(a.Context) > 0 {
		ctxJSON = string(a.Context)
	}

	err := s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		var acked *time.Time
		// prev читається в тому ж знімку, що й UPSERT, тому бачить стан
		// ДО оновлення. Без цього перехід firing→suppressed нічим не
		// відрізнявся б від чергового підтвердження живої проблеми, і UI
		// не дізнався б, що алерт замовк.
		return tx.QueryRow(ctx, `
			WITH prev AS (
				SELECT id, state::text AS old_state
				FROM alr.alerts
				WHERE tenant_id = $1 AND dedup_key = $8
				  AND state IN ('firing','acknowledged','suppressed')
			), ups AS (
				INSERT INTO alr.alerts
					(tenant_id, rule_id, device_id, interface_id, severity, state,
					 title, message, dedup_key, value, threshold, context,
					 suppressed_by, started_at, last_seen_at)
				VALUES ($1, $2, $3, $4, $5::alr.severity,
				        CASE WHEN $12::text IS NULL THEN 'firing' ELSE 'suppressed' END::alr.alert_state,
				        $6, $7, $8, $9, $10, $11::jsonb, $12, now(), now())
				ON CONFLICT (tenant_id, dedup_key)
				WHERE state IN ('firing','acknowledged','suppressed')
				DO UPDATE SET
					last_seen_at = now(),
					value        = EXCLUDED.value,
					-- Підтверджений людиною алерт не повертається у firing
					-- від того, що проблема все ще триває: ack означає
					-- «я знаю», а не «полагоджено».
					state = CASE
						WHEN alr.alerts.state = 'acknowledged' THEN 'acknowledged'
						WHEN $12::text IS NULL THEN 'firing'
						ELSE 'suppressed' END::alr.alert_state,
					suppressed_by = $12,
					severity = EXCLUDED.severity
				RETURNING id, state::text AS new_state, started_at, last_seen_at,
				          notify_count, acked_at, (xmax = 0) AS inserted
			)
			SELECT ups.id::text, ups.new_state, ups.started_at, ups.last_seen_at,
			       ups.notify_count, ups.acked_at, ups.inserted,
			       COALESCE(prev.old_state, '')
			FROM ups LEFT JOIN prev ON prev.id = ups.id
		`, tenantID, nullUUID(a.RuleID), nullUUID(a.DeviceID), nullUUID(a.InterfaceID),
			a.Severity, a.Title, nullString(a.Message), a.DedupKey,
			a.Value, a.Threshold, ctxJSON, nullString(a.SuppressedBy),
		).Scan(&out.ID, &out.State, &out.StartedAt, &out.LastSeenAt,
			&out.NotifyCount, &acked, &out.IsNew, &out.PrevState)
	})
	if err != nil {
		return out, fmt.Errorf("підняття алерту %s: %w", a.DedupKey, err)
	}

	out.RuleID, out.DeviceID, out.InterfaceID = a.RuleID, a.DeviceID, a.InterfaceID
	out.DeviceName, out.Severity, out.Title = a.DeviceName, a.Severity, a.Title
	out.Message, out.DedupKey = a.Message, a.DedupKey
	out.Value, out.Threshold, out.SuppressedBy = a.Value, a.Threshold, a.SuppressedBy
	return out, nil
}

// ResolveMissing закриває алерти правила, яких більше немає серед
// кандидатів.
//
// Закриття робиться саме різницею множин, а не окремою recovery-умовою:
// якщо алерт піднявся тому, що умова виконалась, він має зникнути рівно
// тоді, коли вона перестала. Окрема умова зняття — це другий шанс
// помилитись і залишити на екрані алерт про давно полагоджене.
func (s *Store) ResolveMissing(ctx context.Context, tenantID, ruleID string, keep []string) ([]Alert, error) {
	var out []Alert
	err := s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx, `
			UPDATE alr.alerts
			SET state = 'resolved', resolved_at = now()
			WHERE tenant_id = $1
			  AND rule_id = $2
			  AND state IN ('firing','acknowledged','suppressed')
			  AND NOT (dedup_key = ANY($3::text[]))
			RETURNING id::text, dedup_key, title, severity::text,
			          COALESCE(device_id::text,''), started_at, resolved_at
		`, tenantID, ruleID, keep)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var a Alert
			if err := rows.Scan(&a.ID, &a.DedupKey, &a.Title, &a.Severity,
				&a.DeviceID, &a.StartedAt, &a.ResolvedAt); err != nil {
				return err
			}
			a.State = "resolved"
			out = append(out, a)
		}
		return rows.Err()
	})
	return out, err
}

// ArchiveResolved переносить закриті алерти в гіпертаблицю історії.
//
// Активна таблиця має лишатися маленькою: за неї щосекунди чіпляється
// і мапа, і список алертів, і рахунок непідтверджених. Історія за рік
// у тій самій таблиці зробила б кожен із цих запитів дорожчим без
// жодної користі.
func (s *Store) ArchiveResolved(ctx context.Context, olderThan time.Duration) (int64, error) {
	tag, err := s.bg.Exec(ctx, `
		WITH moved AS (
			DELETE FROM alr.alerts
			WHERE state IN ('resolved','expired')
			  AND resolved_at IS NOT NULL
			  AND resolved_at < now() - $1::interval
			RETURNING *
		)
		INSERT INTO alr.alerts_history
			(ts, id, tenant_id, rule_id, device_id, severity, title,
			 started_at, resolved_at, duration_sec, acked_by, context)
		SELECT COALESCE(resolved_at, now()), id, tenant_id, rule_id, device_id,
		       severity, title, started_at, resolved_at,
		       GREATEST(0, EXTRACT(EPOCH FROM (COALESCE(resolved_at, now()) - started_at))::int),
		       acked_by, context
		FROM moved
		ON CONFLICT (ts, id) DO NOTHING
	`, olderThan.String())
	if err != nil {
		return 0, err
	}
	return tag.RowsAffected(), nil
}

// ---------------------------------------------------------------------
// Придушення
// ---------------------------------------------------------------------

// Suppression — чому саме алерт не має турбувати людину.
type Suppression struct {
	Devices map[string]string // device_id → 'maintenance' | 'mute'
	Rules   map[string]string // rule_id   → те саме
	All     string            // вікно на весь тенант
}

func (s Suppression) For(deviceID, ruleID string) string {
	if s.All != "" {
		return s.All
	}
	if v, ok := s.Devices[deviceID]; ok {
		return v
	}
	if v, ok := s.Rules[ruleID]; ok {
		return v
	}
	return ""
}

// LoadSuppression збирає діючі вікна обслуговування й ручні заглушення.
//
// Читається раз на тік і на весь тенант: перевіряти кожен алерт окремим
// запитом означало б лінійно множити навантаження на БД саме в той
// момент, коли впала половина мережі й алертів найбільше.
func (s *Store) LoadSuppression(ctx context.Context, tenantID string) (Suppression, error) {
	sup := Suppression{Devices: map[string]string{}, Rules: map[string]string{}}

	err := s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx, `
			SELECT selector::text
			FROM alr.maintenance_windows
			WHERE tenant_id = $1 AND period @> now() AND suppress_notifications
		`, tenantID)
		if err != nil {
			return err
		}
		var selectors []string
		for rows.Next() {
			var sel string
			if err := rows.Scan(&sel); err != nil {
				rows.Close()
				return err
			}
			selectors = append(selectors, sel)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return err
		}

		for _, raw := range selectors {
			var sel Selector
			if err := json.Unmarshal([]byte(raw), &sel); err != nil {
				continue
			}
			// Вікно без селектора накриває весь тенант — так виглядає
			// планова робота «вимикаємо все на 30 хвилин».
			if len(sel.DeviceIDs) == 0 && len(sel.GroupIDs) == 0 &&
				len(sel.SiteIDs) == 0 && len(sel.Tags) == 0 {
				sup.All = "maintenance"
				continue
			}
			ids, err := s.resolveSelectorDevices(ctx, tx, tenantID, sel)
			if err != nil {
				return err
			}
			for _, id := range ids {
				sup.Devices[id] = "maintenance"
			}
		}

		rows, err = tx.Query(ctx, `
			SELECT COALESCE(device_id::text,''), COALESCE(rule_id::text,'')
			FROM alr.mutes
			WHERE tenant_id = $1 AND until > now()
		`, tenantID)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var dev, rule string
			if err := rows.Scan(&dev, &rule); err != nil {
				return err
			}
			if dev != "" {
				sup.Devices[dev] = "mute"
			}
			if rule != "" {
				sup.Rules[rule] = "mute"
			}
		}
		return rows.Err()
	})
	return sup, err
}

func (s *Store) resolveSelectorDevices(ctx context.Context, tx pgx.Tx, tenantID string, sel Selector) ([]string, error) {
	a := &args{}
	tenant := a.add(tenantID)
	where, err := s.selectorSQL(sel, a, "d.id")
	if err != nil {
		return nil, err
	}
	q := fmt.Sprintf(`
		SELECT d.id::text FROM inv.devices d
		WHERE d.tenant_id = %s AND d.deleted_at IS NULL %s
	`, tenant, where)

	rows, err := tx.Query(ctx, q, a.vals...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		out = append(out, id)
	}
	return out, rows.Err()
}

// ---------------------------------------------------------------------
// Кореляція за топологією
// ---------------------------------------------------------------------

// TopologyGraph — сусідство пристроїв і їхній поточний стан.
type TopologyGraph struct {
	Neighbors map[string][]string
	Down      map[string]bool
}

// LoadTopology читає граф зв'язків тенанта разом зі станом вузлів.
func (s *Store) LoadTopology(ctx context.Context, tenantID string) (TopologyGraph, error) {
	g := TopologyGraph{Neighbors: map[string][]string{}, Down: map[string]bool{}}

	err := s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx, `
			SELECT id::text, status::text
			FROM inv.devices
			WHERE tenant_id = $1 AND deleted_at IS NULL AND enabled
		`, tenantID)
		if err != nil {
			return err
		}
		for rows.Next() {
			var id, st string
			if err := rows.Scan(&id, &st); err != nil {
				rows.Close()
				return err
			}
			g.Down[id] = st == "down"
			if _, ok := g.Neighbors[id]; !ok {
				g.Neighbors[id] = nil
			}
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return err
		}

		rows, err = tx.Query(ctx, `
			SELECT a_device_id::text, b_device_id::text
			FROM topo.links
			WHERE tenant_id = $1
		`, tenantID)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var a, b string
			if err := rows.Scan(&a, &b); err != nil {
				return err
			}
			g.Neighbors[a] = append(g.Neighbors[a], b)
			g.Neighbors[b] = append(g.Neighbors[b], a)
		}
		return rows.Err()
	})
	return g, err
}

// RootCause визначає, які з недоступних пристроїв є причиною, а які —
// наслідком.
//
// Ідея проста й перевіряється очима на мапі: причина аварії — на межі
// зони недоступності. Пристрій, у якого лишився хоч один живий сусід,
// впав сам; пристрій, оточений виключно мертвими, з великою ймовірністю
// просто стоїть за тим, що впало.
//
// Саме заради цього будувалась topo.links: без кореляції падіння
// маршрутизатора дає сорок сповіщень про сорок «несправних» пристроїв,
// і серед них губиться єдине потрібне.
//
// Повертає device_id → device_id кореневого пристрою. Пристрої-причини
// у мапі відсутні.
func (g TopologyGraph) RootCause() map[string]string {
	collateral := map[string]string{}

	isRoot := func(id string) bool {
		neigh := g.Neighbors[id]
		// Пристрій без жодного відомого зв'язку не можна списати на
		// чужу аварію: топологія про нього нічого не знає, тож єдина
		// чесна відповідь — він упав сам.
		if len(neigh) == 0 {
			return true
		}
		for _, n := range neigh {
			if !g.Down[n] {
				return true
			}
		}
		return false
	}

	// Пошук у ширину від межі: кожен «глибокий» вузол дістає корінь
	// того фронту, через який до нього найкоротший шлях.
	var queue []string
	seen := map[string]bool{}
	for id, down := range g.Down {
		if down && isRoot(id) {
			queue = append(queue, id)
			seen[id] = true
		}
	}

	for len(queue) > 0 {
		cur := queue[0]
		queue = queue[1:]
		root := cur
		if r, ok := collateral[cur]; ok {
			root = r
		}
		for _, n := range g.Neighbors[cur] {
			if !g.Down[n] || seen[n] {
				continue
			}
			seen[n] = true
			collateral[n] = root
			queue = append(queue, n)
		}
	}
	return collateral
}