Netpulse_SasS/server/internal/store/traps.go

package store

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	npv1 "github.com/netpulse/netpulse/gen/go/netpulse/v1"
)

// Приймання й читання SNMP-трапів.
//
// Запис лежить тут, а не в telemetry.go поруч із syslog, з тієї ж
// причини, що й читання аудиту окремо від його запису: класти рядок у
// гіпертаблицю — дешева й незмінна операція, а от що вважати трапом,
// як його назвати й що робити з невідомим відправником — це рішення,
// які житимуть своїм життям.

// ---------------------------------------------------------------------
// Запис
// ---------------------------------------------------------------------

// queueTraps додає в пачку вставки трапів.
//
// varbind-и лягають МАСИВОМ об'єктів, а не мапою «oid → значення», як
// було раніше. Мапа втрачала дві речі, і обидві потрібні:
//
//	тип — `2` в ifOperStatus означає «down», а не число два, і без
//	  ASN.1-типу відрізнити INTEGER від Counter32 у збереженому рядку
//	  вже неможливо;
//	порядок — RFC вимагає, щоб sysUpTime і snmpTrapOID були першими,
//	  і саме порядок varbind-ів у деяких вендорських трапах несе сенс
//	  («перший — той порт, що впав»).
//
// Крім того, мапа мовчки схлопувала повтори: трап із двома ifIndex
// (агрегований лінк) перетворювався на трап з одним.
func queueTraps(batch *pgx.Batch, tenantID string, traps []*npv1.SnmpTrap) {
	for _, t := range traps {
		vbs := make([]storedVarbind, 0, len(t.GetVarbinds()))
		for _, v := range t.GetVarbinds() {
			vbs = append(vbs, storedVarbind{
				OID:   NormalizeOID(v.GetOid()),
				Type:  v.GetType(),
				Value: v.GetValue(),
			})
		}
		payload, err := json.Marshal(vbs)
		if err != nil {
			// Значення вже нормалізовані зондом (недруковане стало
			// hex), тож сюди можна дістатись хіба що новим типом поля.
			// Втратити через це весь трап не можна: сам факт події
			// важливіший за її подробиці.
			payload = []byte("[]")
		}
		batch.Queue(`
			INSERT INTO ts.snmp_traps (ts, tenant_id, device_id, source_ip, trap_oid, varbinds)
			VALUES ($1,$2,$3,$4,$5,$6::jsonb)
		`, tsOrNow(t.GetTs().AsTime()), tenantID, nullUUID(t.GetDeviceId()),
			nullString(t.GetSourceIp()), nullString(NormalizeOID(t.GetTrapOid())), string(payload))
	}
}

// storedVarbind — форма varbind-а в jsonb.
//
// Ключі короткі навмисно: у гіпертаблиці з мільйонами рядків назва
// поля повторюється в кожному з них, і «value» проти «v» — це відчутна
// різниця в розмірі до стиснення.
type storedVarbind struct {
	OID   string `json:"o"`
	Type  string `json:"t,omitempty"`
	Value string `json:"v,omitempty"`
}

// NoteUnknownTrapSources запам'ятовує адреси, які прислали трап, але не
// впізнані як хости.
//
// Навіщо окрема таблиця, коли такі трапи й так лежать у журналі з
// порожнім device_id. Тому що в журналі їх не видно: він відсортований
// за часом, а не за «чи ми це знаємо», і три трапи на добу від чужої
// адреси тонуть між тисячею своїх. А саме ці три — найцінніші: трап від
// адреси, якої немає в інвентарі, зазвичай означає, що в мережі
// з'явилось кероване залізо, про яке моніторинг не знає. Іноді це новий
// комутатор, який забули завести; іноді — чужий пристрій у спільному
// VLAN.
//
// Таблиця має стелю (див. 0065): відправника UDP підробити нічого не
// варте, і без обмеження цей шлях був би способом наростити таблицю
// клієнта з мережі.
func (s *Store) NoteUnknownTrapSources(ctx context.Context, tenantID, agentID string, traps []*npv1.SnmpTrap) error {
	type seen struct {
		count   int
		lastOID string
	}
	byIP := map[string]*seen{}
	for _, t := range traps {
		if t.GetDeviceId() != "" || t.GetSourceIp() == "" {
			continue
		}
		e := byIP[t.GetSourceIp()]
		if e == nil {
			e = &seen{}
			byIP[t.GetSourceIp()] = e
		}
		e.count++
		if oid := NormalizeOID(t.GetTrapOid()); oid != "" {
			e.lastOID = oid
		}
	}
	if len(byIP) == 0 {
		return nil
	}

	return s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		b := &pgx.Batch{}
		for ip, e := range byIP {
			b.Queue(`
				INSERT INTO inv.trap_unknown_sources
					(tenant_id, source_ip, agent_id, first_seen_at, last_seen_at, trap_count, last_trap_oid)
				VALUES ($1, $2::inet, $3, now(), now(), $4, $5)
				ON CONFLICT (tenant_id, source_ip) DO UPDATE SET
					last_seen_at  = now(),
					trap_count    = inv.trap_unknown_sources.trap_count + EXCLUDED.trap_count,
					agent_id      = EXCLUDED.agent_id,
					-- Порожній OID не затирає останній відомий: трап
					-- без ідентифікатора не скасовує того, що ми вже
					-- знали про цього відправника.
					last_trap_oid = COALESCE(NULLIF(EXCLUDED.last_trap_oid, ''),
					                         inv.trap_unknown_sources.last_trap_oid)
			`, tenantID, ip, nullUUID(agentID), e.count, e.lastOID)
		}
		res := tx.SendBatch(ctx, b)
		defer res.Close()
		for i := 0; i < b.Len(); i++ {
			if _, err := res.Exec(); err != nil {
				return fmt.Errorf("невідоме джерело трапа[%d]: %w", i, err)
			}
		}
		return nil
	})
}

// UnknownTrapSource — адреса, яка шле трапи й не є хостом.
type UnknownTrapSource struct {
	SourceIP    string    `json:"source_ip"`
	AgentID     string    `json:"agent_id,omitempty"`
	AgentName   string    `json:"agent_name,omitempty"`
	FirstSeenAt time.Time `json:"first_seen_at"`
	LastSeenAt  time.Time `json:"last_seen_at"`
	TrapCount   int64     `json:"trap_count"`
	LastTrapOID string    `json:"last_trap_oid,omitempty"`
	// Назва останнього трапа, якщо ми її знаємо. Порожня — саме те, що
	// й має бути видно: «прилетіло щось, чого ми не розуміємо, від
	// когось, кого ми не знаємо».
	LastTrapName string `json:"last_trap_name,omitempty"`
}

// ListUnknownTrapSources повертає перелік невпізнаних відправників.
func (s *Store) ListUnknownTrapSources(ctx context.Context, tenantID string) ([]UnknownTrapSource, error) {
	custom, err := s.TrapNames(ctx, tenantID)
	if err != nil {
		return nil, err
	}
	out := []UnknownTrapSource{}
	err = s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx, `
			SELECT host(u.source_ip), COALESCE(u.agent_id::text,''), COALESCE(a.name,''),
			       u.first_seen_at, u.last_seen_at, u.trap_count, COALESCE(u.last_trap_oid,'')
			FROM inv.trap_unknown_sources u
			LEFT JOIN core.agents a ON a.id = u.agent_id
			WHERE u.tenant_id = $1
			ORDER BY u.last_seen_at DESC
			LIMIT 500
		`, tenantID)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var u UnknownTrapSource
			if err := rows.Scan(&u.SourceIP, &u.AgentID, &u.AgentName,
				&u.FirstSeenAt, &u.LastSeenAt, &u.TrapCount, &u.LastTrapOID); err != nil {
				return err
			}
			u.LastTrapName = ResolveTrapOID(custom, u.LastTrapOID).Name
			out = append(out, u)
		}
		return rows.Err()
	})
	return out, err
}

// ForgetUnknownTrapSource прибирає адресу з переліку.
//
// Потрібне не для косметики: перелік — це список питань «а що це таке»,
// і питання, на яке відповіли («це наш новий комутатор, я його завів» /
// «це принтер, і трапи від нього нам не потрібні»), має з нього
// зникати. Список, який не порожнішає, перестають читати.
//
// Адреса, з якої трапи прийдуть знову, з'явиться в переліку знову — і
// це правильно: якщо її так і не завели хостом, питання лишається.
func (s *Store) ForgetUnknownTrapSource(ctx context.Context, tenantID, ip string) error {
	return s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		_, err := tx.Exec(ctx, `
			DELETE FROM inv.trap_unknown_sources WHERE tenant_id = $1 AND source_ip = $2::inet
		`, tenantID, ip)
		return err
	})
}

// ---------------------------------------------------------------------
// Словник кабінету
// ---------------------------------------------------------------------

// TrapNames читає власні відповідності «OID → назва» кабінету.
func (s *Store) TrapNames(ctx context.Context, tenantID string) (map[string]TrapMeaning, error) {
	out := map[string]TrapMeaning{}
	err := s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx, `
			SELECT oid, name, COALESCE(description,'')
			FROM inv.trap_oids WHERE tenant_id = $1
		`, tenantID)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var oid string
			var m TrapMeaning
			if err := rows.Scan(&oid, &m.Name, &m.Description); err != nil {
				return err
			}
			m.Origin = "custom"
			out[oid] = m
		}
		return rows.Err()
	})
	return out, err
}

// SaveTrapName записує або оновлює одну відповідність.
func (s *Store) SaveTrapName(ctx context.Context, tenantID, oid, name, description string) error {
	oid = NormalizeOID(oid)
	if !ValidOID(oid) {
		return fmt.Errorf("%w: %q не схоже на OID — очікуються числа через крапку, "+
			"як 1.3.6.1.4.1.9.9.41.2.0.1", ErrInvalid, oid)
	}
	name = strings.TrimSpace(name)
	if name == "" {
		return fmt.Errorf("%w: назва без тексту нічого не пояснює", ErrInvalid)
	}
	if len([]rune(name)) > 120 {
		return fmt.Errorf("%w: назва довша за 120 символів — це вже опис", ErrInvalid)
	}
	return s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		_, err := tx.Exec(ctx, `
			INSERT INTO inv.trap_oids (tenant_id, oid, name, description, updated_at)
			VALUES ($1,$2,$3,NULLIF($4,''),now())
			ON CONFLICT (tenant_id, oid) DO UPDATE SET
				name = EXCLUDED.name, description = EXCLUDED.description, updated_at = now()
		`, tenantID, oid, name, strings.TrimSpace(description))
		return err
	})
}

// DeleteTrapName прибирає власну відповідність.
//
// Вбудовану так не прибрати, і це навмисно: перекрити її можна (запис
// кабінету має пріоритет), а видалити — ні. Інакше кабінет міг би
// довести себе до стану, у якому linkDown знову показується числом, і
// повернути його було б нічим.
func (s *Store) DeleteTrapName(ctx context.Context, tenantID, oid string) error {
	return s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		_, err := tx.Exec(ctx, `
			DELETE FROM inv.trap_oids WHERE tenant_id = $1 AND oid = $2
		`, tenantID, NormalizeOID(oid))
		return err
	})
}

// ---------------------------------------------------------------------
// Читання журналу трапів
// ---------------------------------------------------------------------

// TrapVarbind — varbind у вигляді, готовому до показу.
type TrapVarbind struct {
	OID   string `json:"oid"`
	Type  string `json:"type,omitempty"`
	Value string `json:"value"`
	// Назва зі словника; порожня означає «стовпець нам невідомий».
	Name string `json:"name,omitempty"`
	// Розшифровка значення там, де її визначає стандарт (up/down).
	ValueLabel string `json:"value_label,omitempty"`
}

// TrapRow — один трап у переліку.
type TrapRow struct {
	ID string    `json:"id"`
	TS time.Time `json:"ts"`

	DeviceID   string `json:"device_id,omitempty"`
	DeviceName string `json:"device_name,omitempty"`
	SourceIP   string `json:"source_ip,omitempty"`

	TrapOID string `json:"trap_oid"`
	// Назва трапа. Порожня — «невідомий трап», і сторінка мусить
	// сказати це словами, а не показати порожню клітинку.
	Name        string `json:"name,omitempty"`
	Description string `json:"description,omitempty"`
	NameOrigin  string `json:"name_origin,omitempty"`

	Varbinds []TrapVarbind `json:"varbinds"`
}

// TrapFilter — питання до журналу трапів.
type TrapFilter struct {
	From time.Time
	To   time.Time

	DeviceIDs []string
	// Адреса або підмережа: '10.20.0.5' і '10.20.0.0/24' обидві
	// працюють через <<=.
	SourceIP string
	OIDs     []string
	// Тільки те, що прийшло від адрес поза інвентарем.
	OnlyUnknown bool
	// Пошук підрядком по значеннях varbind-ів: саме там лежать імена
	// портів і тексти подій.
	Query string

	Limit  int
	Cursor string
}

// TrapPage — сторінка журналу трапів.
type TrapPage struct {
	From  time.Time `json:"from"`
	To    time.Time `json:"to"`
	Traps []TrapRow `json:"traps"`
	// Порожній курсор означає «більше немає»: сторінка береться з
	// limit+1 рядка, тож на це відповідає сама вибірка, а не COUNT по
	// всій гіпертаблиці.
	NextCursor string `json:"next_cursor,omitempty"`
	// Скільки з показаних рядків прийшло від невідомих адрес. Число
	// поруч із переліком, а не окремим запитом: воно потрібне рівно
	// для того, щоб було на що натиснути.
	UnknownInPage int `json:"unknown_in_page"`
}

const (
	trapsDefaultLimit = 50
	trapsMaxLimit     = 200
)

// ListTraps повертає сторінку журналу трапів.
//
// Курсором, а не OFFSET, і з тієї ж причини, що й аудит: гіпертаблиця
// росте вічно, і «двадцята сторінка» через OFFSET означає прочитати
// все, що до неї.
func (s *Store) ListTraps(ctx context.Context, tenantID string, f TrapFilter) (TrapPage, error) {
	limit := f.Limit
	if limit <= 0 {
		limit = trapsDefaultLimit
	}
	if limit > trapsMaxLimit {
		limit = trapsMaxLimit
	}

	custom, err := s.TrapNames(ctx, tenantID)
	if err != nil {
		return TrapPage{}, err
	}

	// Предикат tenant_id стоїть явно й лишається єдиним механізмом
	// ізоляції: ts.snmp_traps — гіпертаблиця, а RLS на ній неможливий,
	// поки ввімкнено стиснення (див. 0011). Прибрати цей рядок означає
	// показати трапи одного кабінету іншому.
	args := []any{tenantID}
	cond := []string{"t.tenant_id = $1"}
	add := func(sql string, v any) {
		args = append(args, v)
		cond = append(cond, fmt.Sprintf(sql, len(args)))
	}

	if f.From.IsZero() {
		f.From = time.Now().Add(-7 * 24 * time.Hour)
	}
	if f.To.IsZero() {
		f.To = time.Now()
	}
	add("t.ts >= $%d", f.From)
	add("t.ts <= $%d", f.To)

	if len(f.DeviceIDs) > 0 {
		add("t.device_id = ANY($%d::uuid[])", f.DeviceIDs)
	}
	if f.OnlyUnknown {
		cond = append(cond, "t.device_id IS NULL")
	}
	if ip := strings.TrimSpace(f.SourceIP); ip != "" {
		add("t.source_ip <<= $%d::inet", ip)
	}
	if len(f.OIDs) > 0 {
		norm := make([]string, 0, len(f.OIDs))
		for _, o := range f.OIDs {
			if n := NormalizeOID(o); n != "" {
				norm = append(norm, n)
			}
		}
		if len(norm) > 0 {
			add("t.trap_oid = ANY($%d::text[])", norm)
		}
	}
	if q := strings.TrimSpace(f.Query); q != "" {
		// Пошук по тексту varbind-ів. Індексу під нього немає й не буде:
		// триграмний GIN на jsonb гіпертаблиці коштував би більше, ніж
		// самі дані, а пошук тут звужений періодом і зазвичай ще й
		// хостом. Ціна видима — широке вікно з пошуком буде повільним.
		add("t.varbinds::text ILIKE '%%' || $%d || '%%'", q)
	}
	if f.Cursor != "" {
		ts, id, err := decodeAuditCursor(f.Cursor)
		if err != nil {
			return TrapPage{}, err
		}
		args = append(args, ts, id)
		cond = append(cond, fmt.Sprintf("(t.ts, t.id) < ($%d, $%d::uuid)", len(args)-1, len(args)))
	}

	args = append(args, limit+1)

	q := `
		SELECT t.ts, t.id::text, COALESCE(t.device_id::text,''), COALESCE(d.name,''),
		       COALESCE(host(t.source_ip),''), COALESCE(t.trap_oid,''), t.varbinds::text
		FROM ts.snmp_traps t
		LEFT JOIN inv.devices d ON d.id = t.device_id
		WHERE ` + strings.Join(cond, " AND ") + `
		ORDER BY t.ts DESC, t.id DESC
		LIMIT $` + strconv.Itoa(len(args))

	// Тенантна транзакція потрібна не самій гіпертаблиці, а JOIN-у на
	// inv.devices: без app.tenant_id політика дала б NULL, і журнал
	// показав би ті самі рядки з порожньою колонкою «хост». Тобто не
	// порожній екран, а правдоподібну неправду — «усі трапи прийшли
	// нізвідки».
	out := TrapPage{From: f.From, To: f.To, Traps: []TrapRow{}}
	err = s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx, q, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var r TrapRow
			var raw string
			if err := rows.Scan(&r.TS, &r.ID, &r.DeviceID, &r.DeviceName,
				&r.SourceIP, &r.TrapOID, &raw); err != nil {
				return err
			}
			m := ResolveTrapOID(custom, r.TrapOID)
			r.Name, r.Description, r.NameOrigin = m.Name, m.Description, m.Origin
			r.Varbinds = decodeVarbinds(custom, raw)
			out.Traps = append(out.Traps, r)
		}
		return rows.Err()
	})
	if err != nil {
		return TrapPage{}, err
	}

	if len(out.Traps) > limit {
		last := out.Traps[limit-1]
		out.Traps = out.Traps[:limit]
		out.NextCursor = encodeAuditCursor(last.TS, last.ID)
	}
	for _, r := range out.Traps {
		if r.DeviceID == "" {
			out.UnknownInPage++
		}
	}
	return out, nil
}

// decodeVarbinds читає збережений jsonb.
//
// Двома формами навмисно. До цієї гілки varbind-и лягали мапою
// «oid → значення», і в базі клієнтів такі рядки лежать. Читати їх як
// помилку означало б, що весь накопичений архів трапів зник у той
// вечір, коли викотили оновлення, — і зник саме тоді, коли вперше
// з'явилась сторінка, на якій його видно.
func decodeVarbinds(custom map[string]TrapMeaning, raw string) []TrapVarbind {
	out := []TrapVarbind{}
	if strings.TrimSpace(raw) == "" {
		return out
	}

	var arr []storedVarbind
	if err := json.Unmarshal([]byte(raw), &arr); err == nil {
		for _, v := range arr {
			out = append(out, enrichVarbind(custom, v.OID, v.Type, v.Value))
		}
		return out
	}

	var flat map[string]string
	if err := json.Unmarshal([]byte(raw), &flat); err == nil {
		keys := make([]string, 0, len(flat))
		for k := range flat {
			keys = append(keys, k)
		}
		// Порядок мапи в Go випадковий, а на екрані рядки не мають
		// стрибати між оновленнями сторінки. Для старої форми порядку
		// не існує взагалі, тож беремо будь-який СТАЛИЙ.
		slices.Sort(keys)
		for _, k := range keys {
			out = append(out, enrichVarbind(custom, k, "", flat[k]))
		}
	}
	return out
}

func enrichVarbind(custom map[string]TrapMeaning, oid, typ, value string) TrapVarbind {
	oid = NormalizeOID(oid)
	return TrapVarbind{
		OID:        oid,
		Type:       typ,
		Value:      value,
		Name:       ResolveVarbindOID(custom, oid),
		ValueLabel: DescribeVarbindValue(oid, value),
	}
}