Netpulse_SasS/server/internal/store/autochecks.go

package store

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

	"github.com/jackc/pgx/v5"
	npv1 "github.com/netpulse/netpulse/gen/go/netpulse/v1"
	"google.golang.org/protobuf/types/known/durationpb"
)

// Скільки інтерфейсів максимум класти в один snmp.if-чек.
//
// Модуль б'є запити порціями по 24 змінні, а на кожен порт припадає 10
// OID. 256 портів — це вже 107 PDU за один цикл опитування; далі чек
// перестає вкладатись у власний таймаут раніше, ніж у ліміти пристрою.
// Комутатори з більшою кількістю портів треба ділити на кілька чеків —
// поки що просто обрізаємо й пишемо про це в журнал.
const MaxInterfacesPerCheck = 256

// Скільки рядків inv.interfaces узагалі читати за раз.
//
// Стеля вище — про розмір PDU й лишається недоторканою. Ця — про те, що
// фільтр складу портів застосовує Go, а отже кандидатів треба спершу
// прочитати: LIMIT 257, який стояв тут раніше, відрізав би порти ще до
// того, як зразок їх побачив, і «xe-0/0/47» просто не існувало б для
// фільтра. Число з великим запасом: найбільша коробка на стенді має 100
// портів, шасі з повною набивкою — близько тисячі.
const maxInterfaceScan = 4096

// ifEligibleSQL — базові правила складу портів, спільні для чека й
// попереднього перегляду.
//
// Одним рядком, а не двома копіями, саме тому, що перегляд мусить
// показувати те, що станеться насправді. Розійшовшись на одному «і»,
// два переліки почали б відрізнятись рівно на ті порти, через які
// людина й прийшла дивитись.
//
// Loopback і відсутні порти графіка не дають, а місце в PDU займають;
// `monitored` — ручний вимикач людини на конкретному порту, і фільтр
// шаблону його не скасовує.
const ifEligibleSQL = `monitored
	  AND if_index IS NOT NULL
	  AND COALESCE(type, '') <> 'softwareLoopback'
	  AND oper_status <> 'notPresent'`

// InterfaceCheckInterval — типовий інтервал опитування лічильників.
//
// 60 секунд — компроміс: частіше не має сенсу для 32-бітних лічильників
// на повільних каналах, рідше — анімація трафіку на мапі стає слайдшоу.
const InterfaceCheckInterval = 60 * time.Second

// ifCheckParams — те, що лягає в core.checks.params для snmp.if.
type ifCheckParams struct {
	UseHCCounters bool            `json:"use_hc_counters"`
	Interfaces    []ifCheckTarget `json:"interfaces"`
}

type ifCheckTarget struct {
	IfIndex     int64  `json:"if_index"`
	InterfaceID string `json:"interface_id"`
	SpeedBps    uint64 `json:"speed_bps"`
}

// EnsureInterfaceChecks створює або оновлює snmp.if-чек для пристрою за
// поточним вмістом inv.interfaces.
//
// Навіщо це на сервері, а не на агенті: агент не має права вирішувати,
// що опитувати — це впирається в ліміти тарифу й у те, які інтерфейси
// оператор позначив як непотрібні. Агент лише виконує список.
//
// Повертає задачу для TaskDelta, якщо щось змінилось. nil означає
// «нічого робити»: або немає SNMP-креденшела, або немає інтерфейсів,
// або список не змінився з минулого разу.
//
// Чому це й далі Go, а не прототип шаблону (0059). Порт — теж рядок
// таблиці зі змінним індексом, і спокуса виразити його прототипом
// очевидна. Але snmp.if не збирає метрики: він повертає
// InterfaceCounters, а для цього зонд тримає попередній замір, рахує
// швидкості за ФАКТИЧНИМ інтервалом, ловить перевертання лічильника й
// ділить на номінальну швидкість порту. Прототип віддав би сирі октети
// без жодного з цих чотирьох умінь.
//
// Далі — прив'язки. Лічильники лягають у ts.if_counters за
// interface_id, а не в ts.samples за міткою; на цьому interface_id
// тримаються анімація трафіку на мапі, інспектор лінка, filtered_out_at
// і тригери з джерелом «interface». Прототип породжує серії з мітками й
// без interface_id.
//
// Тобто спільного між ними рівно одне слово — «таблиця». Прототипи
// покривають випадок, якого раніше не було покрито нічим; цей випадок
// покритий, і переносити його заради симетрії означало б обміняти
// найпомітнішу частину продукту на мінус двісті рядків.
func (s *Store) EnsureInterfaceChecks(ctx context.Context, a *Agent, deviceID string) (*npv1.Task, error) {
	var task *npv1.Task

	err := s.InTenantTx(ctx, a.TenantID, func(tx pgx.Tx) error {
		// Без SNMP-креденшела чек лише щохвилини писав би помилку
		// автентифікації — це шум, а не моніторинг.
		var hasCred bool
		if err := tx.QueryRow(ctx, `
			SELECT EXISTS (
				SELECT 1 FROM inv.device_credentials dc
				JOIN inv.credentials c ON c.id = dc.credential_id
				WHERE dc.device_id = $1
				  AND c.tenant_id = $2
				  AND c.proto IN ('snmp_v2c','snmp_v3')
			)
		`, deviceID, a.TenantID).Scan(&hasCred); err != nil {
			return err
		}
		if !hasCred {
			return nil
		}

		targets, err := selectInterfaceTargets(ctx, tx, a.TenantID, deviceID)
		if err != nil {
			return err
		}
		params := ifCheckParams{UseHCCounters: true, Interfaces: targets}

		truncated := false
		if len(params.Interfaces) > MaxInterfacesPerCheck {
			params.Interfaces = params.Interfaces[:MaxInterfacesPerCheck]
			truncated = true
		}
		if len(params.Interfaces) == 0 {
			return nil
		}

		payload, err := json.Marshal(params)
		if err != nil {
			return err
		}

		// Шукаємо існуючий чек будь-яких параметрів: унікальний індекс
		// core.checks включає md5(params), тому наївний upsert плодив би
		// новий рядок на кожну зміну складу портів.
		var (
			checkID  string
			oldJSON  string
			interval int32
			timeout  int32
			retries  int32
		)
		err = tx.QueryRow(ctx, `
			SELECT id::text, params::text, interval_sec, timeout_ms, retries
			FROM core.checks
			WHERE device_id = $1 AND tenant_id = $2 AND check_type = 'snmp.if'
			ORDER BY created_at
			LIMIT 1
		`, deviceID, a.TenantID).Scan(&checkID, &oldJSON, &interval, &timeout, &retries)

		switch {
		case err == nil:
			if sameInterfaceSet(oldJSON, payload) {
				return nil
			}
			if _, err := tx.Exec(ctx, `
				UPDATE core.checks
				SET params = $2::jsonb, enabled = true, updated_at = now()
				WHERE id = $1
			`, checkID, string(payload)); err != nil {
				return err
			}

		case errors.Is(err, pgx.ErrNoRows):
			interval = int32(InterfaceCheckInterval / time.Second)
			timeout, retries = 15000, 1
			if err := tx.QueryRow(ctx, `
				INSERT INTO core.checks
					(tenant_id, device_id, check_type, params, interval_sec, timeout_ms, retries)
				VALUES ($1, $2, 'snmp.if', $3::jsonb, $4, $5, $6)
				RETURNING id::text
			`, a.TenantID, deviceID, string(payload), interval, timeout, retries).Scan(&checkID); err != nil {
				return err
			}

		default:
			return err
		}

		if truncated {
			return fmt.Errorf("пристрій %s має понад %d інтерфейсів: чек обрізано",
				deviceID, MaxInterfacesPerCheck)
		}

		intervalDur := time.Duration(interval) * time.Second
		task = &npv1.Task{
			CheckId:        checkID,
			DeviceId:       deviceID,
			CheckType:      "snmp.if",
			ParamsJson:     payload,
			Interval:       durationpb.New(intervalDur),
			Timeout:        durationpb.New(time.Duration(timeout) * time.Millisecond),
			Retries:        uint32(retries),
			Enabled:        true,
			ScheduleOffset: durationpb.New(ScheduleOffset(checkID, intervalDur)),
		}
		return nil
	})

	return task, err
}

// sameInterfaceSet порівнює склад портів, ігноруючи порядок ключів у JSON.
//
// Пряме порівняння рядків давало б хибну зміну щоразу, коли Postgres
// інакше впорядкує ключі jsonb, і агент отримував би новий план на
// кожен обхід автовиявлення.
func sameInterfaceSet(oldJSON string, newJSON []byte) bool {
	var a, b ifCheckParams
	if err := json.Unmarshal([]byte(oldJSON), &a); err != nil {
		return false
	}
	if err := json.Unmarshal(newJSON, &b); err != nil {
		return false
	}
	if a.UseHCCounters != b.UseHCCounters || len(a.Interfaces) != len(b.Interfaces) {
		return false
	}

	seen := make(map[string]ifCheckTarget, len(a.Interfaces))
	for _, t := range a.Interfaces {
		seen[t.InterfaceID] = t
	}
	for _, t := range b.Interfaces {
		prev, ok := seen[t.InterfaceID]
		if !ok || prev.IfIndex != t.IfIndex || prev.SpeedBps != t.SpeedBps {
			return false
		}
	}
	return true
}

// selectInterfaceTargets вирішує, які порти хоста потраплять у чек.
//
// Одне місце на два виклики (перебудова після автовиявлення й
// перебудова після зміни шаблонів) — саме тому, що правило «який порт
// опитувати» мусить бути одне. Два його примірники розійшлися б, і
// склад чека залежав би від того, що саме щойно сталося.
//
// Заразом розставляє позначки `filtered_out_at` на самих портах: без
// них обрив графіка нічим не відрізнити від аварії.
func selectInterfaceTargets(ctx context.Context, tx pgx.Tx, tenantID, deviceID string) ([]ifCheckTarget, error) {
	// Фільтри читаємо ДО портів: якщо серед них є зіпсований, склад
	// чека краще не чіпати взагалі, ніж перебудувати його за половиною
	// правил.
	set, err := deviceInterfaceFilters(ctx, tx, deviceID)
	if err != nil {
		return nil, err
	}

	rows, err := tx.Query(ctx, `
		SELECT if_index, id::text, COALESCE(speed_bps, 0),
		       name, COALESCE(alias, ''), COALESCE(type, '')
		FROM inv.interfaces
		WHERE device_id = $1
		  AND tenant_id = $2
		  AND `+ifEligibleSQL+`
		ORDER BY if_index
		LIMIT $3
	`, deviceID, tenantID, maxInterfaceScan)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var (
		targets []ifCheckTarget
		// Порти, які пройшли базові правила, але не пройшли фільтр.
		excluded []string
		eligible int
	)
	for rows.Next() {
		var t ifCheckTarget
		var in InterfaceCandidate
		if err := rows.Scan(&t.IfIndex, &t.InterfaceID, &t.SpeedBps,
			&in.Name, &in.Alias, &in.Type); err != nil {
			return nil, err
		}
		eligible++
		if set.match(in) {
			targets = append(targets, t)
		} else {
			excluded = append(excluded, t.InterfaceID)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	// Фільтр, який не пропустив НІЧОГО, — це майже завжди помилка в
	// зразку, а не намір: порожній snmp.if-чек ззовні не відрізняється
	// від справного, поки хтось не помітить, що метрик портів немає вже
	// тиждень.
	//
	// Тому склад чека лишається таким, який був, а транзакція
	// відкочується цілком — разом із позначками на портах. Викличник
	// пише це в журнал, а людина бачить те саме число в попередньому
	// перегляді ще до збереження.
	if !set.empty() && eligible > 0 && len(targets) == 0 {
		return nil, fmt.Errorf("%w: пристрій %s, портів під наглядом %d, фільтр шаблонів (%s) не пропустив жодного — склад чека лишено без змін",
			ErrInterfaceFilterEmpty, deviceID, eligible, strings.Join(set.names, ", "))
	}

	if err := markFilteredOut(ctx, tx, tenantID, deviceID, excluded); err != nil {
		return nil, err
	}
	return targets, nil
}

// ApplyInterfaceFilters переписує склад уже наявного snmp.if-чека.
//
// Потрібна тоді, коли змінились не порти, а правила: шаблон причепили,
// відчепили або поправили в ньому зразок. Чекати наступного обходу
// автовиявлення означало б, що людина натиснула «Зберегти» й нічого не
// сталося — а «нічого» тут може тривати годинами.
//
// Чека НЕ створює навмисно: створення прив'язане до звіту
// автовиявлення, бо лише там відомо, який агент опитує цей хост і кому
// штовхати оновлений план. Тут ми лише приводимо у відповідність те, що
// вже є; агент забере новий склад разом із наступною звіркою плану.
func (s *Store) ApplyInterfaceFilters(ctx context.Context, tenantID, deviceID string) error {
	return s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		return applyInterfaceFiltersTx(ctx, tx, tenantID, deviceID)
	})
}

func applyInterfaceFiltersTx(ctx context.Context, tx pgx.Tx, tenantID, deviceID string) error {
	var checkID string
	if err := tx.QueryRow(ctx, `
		SELECT id::text FROM core.checks
		WHERE device_id = $1 AND tenant_id = $2 AND check_type = 'snmp.if'
		ORDER BY created_at LIMIT 1
	`, deviceID, tenantID).Scan(&checkID); err != nil {
		if isNoRows(err) {
			return nil
		}
		return err
	}

	targets, err := selectInterfaceTargets(ctx, tx, tenantID, deviceID)
	switch {
	case errors.Is(err, ErrInterfaceFilterEmpty):
		// Тут ця відмова не має нікуди підійматись. Реконсиляція
		// шаблонів іде і в потоці розпізнавання хоста, і в масовій
		// правці — там помилка одного фільтра поклала б увесь звіт, а
		// цей самий факт і так буде сказаний тричі: 409 при збереженні
		// шаблону, попередження в журналі автовиявлення й число «0 зі
		// 100» у попередньому перегляді.
		//
		// Склад чека при цьому лишається попереднім — саме те, чого ми
		// й хочемо: перестати збирати все відразу гірше, ніж збирати
		// зайве.
		return nil
	case err != nil:
		return err
	}
	if len(targets) > MaxInterfacesPerCheck {
		targets = targets[:MaxInterfacesPerCheck]
	}
	// Порожній склад чека не пишемо ніколи: або фільтр помилковий (і
	// тоді вище вже була помилка), або портів справді не лишилось — і
	// тоді чек має дочекатись автовиявлення, яке одне знає, чи це
	// назавжди.
	if len(targets) == 0 {
		return nil
	}

	payload, err := json.Marshal(ifCheckParams{UseHCCounters: true, Interfaces: targets})
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `
		UPDATE core.checks SET params = $2::jsonb, updated_at = now()
		WHERE id = $1
	`, checkID, string(payload))
	return err
}

// ErrInterfaceFilterEmpty — фільтри шаблонів не пропустили жодного порту.
//
// Окрема помилка, а не просто текст: її ловлять і сервер (щоб не
// перебудовувати чек), і HTTP-шар (щоб відмовити в збереженні шаблону
// зрозумілим кодом, а не «внутрішньою помилкою»).
var ErrInterfaceFilterEmpty = errors.New("фільтр портів не пропустив жодного порту")

// deviceInterfaceFilters збирає фільтри всіх шаблонів хоста.
//
// Шаблони без фільтра пропускаються: відсутність думки про порти не є
// думкою «брати все» — інакше один шаблон загального призначення
// знецінював би фільтр іншого.
//
// Зіпсований фільтр (не компілюється) — це помилка, а не «беремо все».
// Розширити склад чека зі ста портів назад до ста через друкарську
// помилку в чужому шаблоні гірше, ніж лишити все як є й написати про це.
func deviceInterfaceFilters(ctx context.Context, tx pgx.Tx, deviceID string) (*interfaceFilterSet, error) {
	rows, err := tx.Query(ctx, `
		SELECT t.name, t.if_filter::text
		FROM tpl.device_templates dt
		JOIN tpl.templates t ON t.id = dt.template_id
		WHERE dt.device_id = $1
		ORDER BY t.name
	`, deviceID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	set := &interfaceFilterSet{}
	for rows.Next() {
		var name, raw string
		if err := rows.Scan(&name, &raw); err != nil {
			return nil, err
		}
		f := parseInterfaceFilter(raw)
		if f.IsEmpty() {
			continue
		}
		cf, err := compileInterfaceFilter(f)
		if err != nil {
			return nil, fmt.Errorf("фільтр портів шаблону «%s»: %w", name, err)
		}
		set.names = append(set.names, name)
		set.filters = append(set.filters, cf)
	}
	return set, rows.Err()
}

// markFilteredOut ставить і знімає позначку «виключено фільтром».
//
// Двома запитами, а не одним UPDATE по всіх портах: обидва мають умову
// на поточне значення, тож порт, стан якого не змінився, не оновлюється
// взагалі. Це не мікрооптимізація — на кожному порту висить тригер
// touch_updated_at, і сліпий UPDATE ворушив би `updated_at` сотні портів
// щохвилини, роблячи це поле марним.
func markFilteredOut(ctx context.Context, tx pgx.Tx, tenantID, deviceID string, excluded []string) error {
	if excluded == nil {
		excluded = []string{}
	}
	if _, err := tx.Exec(ctx, `
		UPDATE inv.interfaces SET filtered_out_at = now()
		WHERE device_id = $1 AND tenant_id = $2
		  AND id = ANY($3::uuid[])
		  AND filtered_out_at IS NULL
	`, deviceID, tenantID, excluded); err != nil {
		return err
	}
	_, err := tx.Exec(ctx, `
		UPDATE inv.interfaces SET filtered_out_at = NULL
		WHERE device_id = $1 AND tenant_id = $2
		  AND NOT (id = ANY($3::uuid[]))
		  AND filtered_out_at IS NOT NULL
	`, deviceID, tenantID, excluded)
	return err
}

// PlanHash перераховує хеш плану без побудови самого плану.
//
// Потрібен після зміни чеків: агент має отримати новий хеш разом із
// дельтою, інакше після реконекту він доповість старий, сервер вирішить,
// що план застарів, і перезаллє все повністю.
func (s *Store) PlanHash(ctx context.Context, a *Agent) ([]byte, error) {
	plan, err := s.BuildPlan(ctx, a)
	if err != nil {
		return nil, err
	}
	return plan.GetPlanHash(), nil
}

// Два інтервали, бо в чека розпізнавання дві різні ролі.
//
// Поки хост не розпізнано, від цього чека залежить усе інше: підбір
// шаблонів, підбір профілю збору конфігів. Людина щойно завела хост і
// дивиться на екран — тут дорога кожна хвилина.
//
// Щойно розпізнано, питання закрите й лишається тільки сторожова
// функція: залізо в шафі міняють, і картка має помітити це сама.
// Три OID раз на шість годин — нижче за рівень шуму на будь-якому
// пристрої.
//
// Розклад агента кладе задачі на сітку інтервалу: при шести годинах
// перший запуск міг би припасти й на п'яту годину очікування. Саме
// тому початковий інтервал малий, а не тому, що дані швидко старіють.
const (
	IdentifyProbeInterval = 2 * time.Minute
	IdentifyCheckInterval = 6 * time.Hour
)

// IdentifyCheckType — тип чека розпізнавання.
//
// Власний тип, а не полегшений `topology.discover`, хоч модуль той
// самий: у списку перевірок хоста рядок мусить називати те, що робить.
// «Neighbor Discovery», який не шукає жодного сусіда, — це підказка,
// що бреше.
//
// Префікс `topology.` обов'язковий: Registry.Register вимагає, щоб тип
// чека починався з ключа модуля.
const IdentifyCheckType = "topology.identify"

// EnsureIdentifyChecks заводить чек розпізнавання хостам, які його
// потребують.
//
// Підбір профілю збору конфігів і шаблонів опитування спирається на
// sysObjectID — а він нізвідки не береться, доки хост нічим не
// опитують. Виходило замкнене коло: автоматика чекає на дані, дані
// чекають на те, що людина руками заведе чек автовиявлення.
//
// Розриваємо його тут: хост має SNMP-доступ — значить, дозволено його
// запитати, хто він такий.
//
// Чек не створюється, якщо автовиявлення вже налаштоване вручну: його
// системна група так само заповнить поля, а два чеки одного типу лише
// подвоїли б трафік.
func (s *Store) EnsureIdentifyChecks(ctx context.Context, a *Agent) (int, error) {
	var made int
	err := s.InTenantTx(ctx, a.TenantID, func(tx pgx.Tx) error {
		// Приводимо інтервал до ролі: нерозпізнаним — часто,
		// розпізнаним — рідко. В обидва боки, бо пристрій може і
		// зникнути з обліку, і повернутись іншою моделлю.
		if _, err := tx.Exec(ctx, `
			UPDATE core.checks ch
			SET interval_sec = want.sec, updated_at = now()
			FROM inv.devices d,
			     LATERAL (SELECT CASE WHEN d.sys_object_id IS NULL
			                          THEN $3::int ELSE $4::int END) AS want(sec)
			WHERE d.id = ch.device_id
			  AND ch.tenant_id = $1
			  AND d.agent_id = $2
			  AND ch.check_type = $5
			  AND ch.interval_sec <> want.sec
		`, a.TenantID, a.ID,
			int32(IdentifyProbeInterval/time.Second),
			int32(IdentifyCheckInterval/time.Second),
			IdentifyCheckType); err != nil {
			return err
		}

		// Чек заводиться будь-якому хосту зі SNMP-доступом, а не лише
		// нерозпізнаному.
		//
		// Спершу умова була «ще не розпізнаний»: чек потрібен, щоб
		// хост розпізнати, а далі він просто лишається. Хибність цього
		// міркування показав живий стенд: коли чек зникав з іншої
		// причини, він не повертався НІКОЛИ — хост уже розпізнаний, під
		// умову не підпадає, і сторожова функція (помітити заміну
		// заліза) мовчки припинялась назавжди.
		//
		// Різниця між «розпізнати» і «стежити» лишилась там, де їй
		// місце, — в інтервалі вище: дві хвилини проти шести годин.
		return tx.QueryRow(ctx, `
			WITH candidates AS (
				SELECT d.id
				FROM inv.devices d
				WHERE d.tenant_id = $1
				  AND d.agent_id = $2
				  AND d.enabled
				  AND d.deleted_at IS NULL
				  AND EXISTS (
					SELECT 1 FROM inv.device_credentials dc
					JOIN inv.credentials c ON c.id = dc.credential_id
					WHERE dc.device_id = d.id
					  AND c.tenant_id = d.tenant_id
					  AND c.proto IN ('snmp_v2c','snmp_v3')
				  )
				  -- Ні свого чека, ні вручну заведеного автовиявлення:
				  -- воно знімає ту саму системну групу, і другий чек
				  -- лише подвоїв би трафік.
				  AND NOT EXISTS (
					SELECT 1 FROM core.checks ch
					WHERE ch.device_id = d.id
					  AND ch.tenant_id = d.tenant_id
					  AND ch.check_type IN ($3, 'topology.discover')
				  )
			), ins AS (
				INSERT INTO core.checks
					(tenant_id, device_id, check_type, params, interval_sec, timeout_ms, retries)
				SELECT $1, candidates.id, $3, '{}'::jsonb, $4, 10000, 1
				FROM candidates
				RETURNING 1
			)
			SELECT count(*) FROM ins
		`, a.TenantID, a.ID, IdentifyCheckType, int32(IdentifyProbeInterval/time.Second)).Scan(&made)
	})
	return made, err
}