Netpulse_SasS/server/internal/store/discovery.go

package store

import (
	"context"
	"encoding/json"
	"errors"
	"strconv"
	"strings"

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

// Рівні впевненості зіставлення сусіда з інвентарем.
//
// Порядок не довільний: chassis-id унікальний за стандартом LLDP,
// MAC — майже, IP керування може повторюватись у різних VRF, а
// sysName взагалі вводить людина й на двох комутаторах цілком може
// бути "switch". Тому лінк, зведений за іменем, отримує низьку
// впевненість і не має мовчки заміщати те, що підтвердила людина.
const (
	confChassis = 95
	confMAC     = 90
	confMgmtIP  = 80
	confSysName = 60
)

// DiscoveryStats — підсумок обробки звіту, їде в DiscoveryAck.
type DiscoveryStats struct {
	NeighborsSeen     int
	NeighborsResolved int
	LinksCreated      int
	LinksUpdated      int
	InterfacesUpdated int
}

// ApplyDiscovery приймає звіт агента: зберігає сирих сусідів,
// зіставляє їх з інвентарем і зводить у topo.links.
//
// Агент не вирішує, хто з ким з'єднаний, — він доповідає лише
// «на порту X бачу chassis Y, port Z». Уся інтерпретація тут, тому
// правила зіставлення можна міняти без оновлення зондів у полі.
func (s *Store) ApplyDiscovery(ctx context.Context, a *Agent, rep *npv1.DiscoveryReport) (DiscoveryStats, error) {
	var st DiscoveryStats

	if err := s.upsertInterfaces(ctx, a, rep.GetInterfaces(), &st); err != nil {
		return st, err
	}

	for _, n := range rep.GetNeighbors() {
		st.NeighborsSeen++
		if err := s.applyNeighbor(ctx, a, n, &st); err != nil {
			return st, err
		}
	}

	return st, nil
}

func (s *Store) upsertInterfaces(ctx context.Context, a *Agent, ifs []*npv1.InterfaceRecord, st *DiscoveryStats) error {
	if len(ifs) == 0 {
		return nil
	}

	return s.InTenantTx(ctx, a.TenantID, func(tx pgx.Tx) error {
		for _, r := range ifs {
			_, err := tx.Exec(ctx, `
				INSERT INTO inv.interfaces
					(tenant_id, device_id, if_index, name, alias, mac, mtu, type,
					 speed_bps, duplex, admin_status, oper_status)
				VALUES ($1,$2,$3,$4,$5,$6::macaddr,$7,$8,$9,$10,
				        $11::inv.if_admin_status, $12::inv.if_oper_status)
				ON CONFLICT (device_id, if_index) WHERE if_index IS NOT NULL
				DO UPDATE SET
					name = EXCLUDED.name,
					alias = EXCLUDED.alias,
					mac = COALESCE(EXCLUDED.mac, inv.interfaces.mac),
					speed_bps = COALESCE(NULLIF(EXCLUDED.speed_bps, 0), inv.interfaces.speed_bps),
					admin_status = EXCLUDED.admin_status,
					oper_status = EXCLUDED.oper_status,
					updated_at = now()
			`, a.TenantID, r.GetDeviceId(), r.GetIfIndex(), r.GetName(),
				nullString(r.GetAlias()), nullString(r.GetMac()),
				nullInt(int(r.GetMtu())), nullString(r.GetType()),
				nullInt64(int64(r.GetSpeedBps())), nullString(r.GetDuplex()),
				statusOr(r.GetAdminStatus(), "unknown"),
				statusOr(r.GetOperStatus(), "unknown"))
			if err != nil {
				return err
			}
			st.InterfacesUpdated++
		}
		return nil
	})
}

func (s *Store) applyNeighbor(ctx context.Context, a *Agent, n *npv1.NeighborRecord, st *DiscoveryStats) error {
	return s.InTenantTx(ctx, a.TenantID, func(tx pgx.Tx) error {
		localIfID, err := s.resolveLocalInterface(ctx, tx, a.TenantID, n)
		if err != nil {
			return err
		}

		remoteDevID, conf, err := s.resolveRemoteDevice(ctx, tx, a.TenantID, n)
		if err != nil {
			return err
		}

		var remoteIfID any
		if remoteDevID != nil {
			remoteIfID, err = s.resolveRemoteInterface(ctx, tx, a.TenantID, *remoteDevID, n.GetRemotePortId(), n.GetRemotePortDescr())
			if err != nil {
				return err
			}
		}

		raw := n.GetRawJson()
		if len(raw) == 0 {
			raw = []byte("{}")
		} else if !json.Valid(raw) {
			raw = []byte("{}")
		}

		_, err = tx.Exec(ctx, `
			INSERT INTO topo.neighbors
				(tenant_id, device_id, interface_id, proto,
				 remote_chassis_id, remote_system_name, remote_port_id, remote_port_descr,
				 remote_mgmt_ip, remote_mac, remote_platform, remote_capabilities,
				 resolved_device_id, resolved_interface_id, confidence, last_seen_at, raw)
			VALUES ($1,$2,$3,$4::topo.discovery_proto,$5,$6,$7,$8,
			        $9::inet,$10::macaddr,$11,$12,$13,$14,$15,now(),$16::jsonb)
			ON CONFLICT (device_id, COALESCE(interface_id, '00000000-0000-0000-0000-000000000000'::uuid),
			             proto, COALESCE(remote_chassis_id,''), COALESCE(remote_port_id,''),
			             COALESCE(remote_mac::text,''))
			DO UPDATE SET
				remote_system_name = EXCLUDED.remote_system_name,
				remote_port_descr = EXCLUDED.remote_port_descr,
				remote_mgmt_ip = EXCLUDED.remote_mgmt_ip,
				remote_mac = EXCLUDED.remote_mac,
				remote_platform = EXCLUDED.remote_platform,
				remote_capabilities = EXCLUDED.remote_capabilities,
				resolved_device_id = EXCLUDED.resolved_device_id,
				resolved_interface_id = EXCLUDED.resolved_interface_id,
				confidence = EXCLUDED.confidence,
				last_seen_at = now(),
				raw = EXCLUDED.raw
		`, a.TenantID, n.GetDeviceId(), localIfID, protoName(n.GetProto()),
			nullString(n.GetRemoteChassisId()), nullString(n.GetRemoteSystemName()),
			nullString(n.GetRemotePortId()), nullString(n.GetRemotePortDescr()),
			nullString(n.GetRemoteMgmtIp()), nullString(n.GetRemoteMac()),
			nullString(n.GetRemotePlatform()), n.GetRemoteCapabilities(),
			remoteDevID, remoteIfID, conf, string(raw))
		if err != nil {
			return err
		}

		if remoteDevID == nil {
			return nil
		}
		st.NeighborsResolved++

		created, err := s.upsertLink(ctx, tx, a.TenantID, linkSpec{
			aDevice:    n.GetDeviceId(),
			aInterface: localIfID,
			bDevice:    *remoteDevID,
			bInterface: remoteIfID,
			proto:      protoName(n.GetProto()),
			confidence: conf,
		})
		if err != nil {
			return err
		}
		if created {
			st.LinksCreated++
		} else {
			st.LinksUpdated++
		}
		return nil
	})
}

func (s *Store) resolveLocalInterface(ctx context.Context, tx pgx.Tx, tenantID string, n *npv1.NeighborRecord) (any, error) {
	if n.GetLocalInterfaceId() != "" {
		return n.GetLocalInterfaceId(), nil
	}

	var id string
	var err error
	switch {
	case n.GetLocalIfIndex() > 0:
		err = tx.QueryRow(ctx, `
			SELECT id::text FROM inv.interfaces
			WHERE tenant_id = $1 AND device_id = $2 AND if_index = $3
		`, tenantID, n.GetDeviceId(), n.GetLocalIfIndex()).Scan(&id)
	case n.GetLocalPortName() != "":
		err = tx.QueryRow(ctx, `
			SELECT id::text FROM inv.interfaces
			WHERE tenant_id = $1 AND device_id = $2 AND lower(name) = lower($3)
		`, tenantID, n.GetDeviceId(), n.GetLocalPortName()).Scan(&id)
	default:
		return nil, nil
	}

	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return id, nil
}

// resolveRemoteDevice шукає пристрій за спаданням надійності ознаки.
func (s *Store) resolveRemoteDevice(ctx context.Context, tx pgx.Tx, tenantID string, n *npv1.NeighborRecord) (*string, int, error) {
	type probe struct {
		query string
		arg   string
		conf  int
	}

	probes := []probe{
		{`SELECT id::text FROM inv.devices
		    WHERE tenant_id=$1 AND deleted_at IS NULL AND chassis_id = $2 LIMIT 1`,
			n.GetRemoteChassisId(), confChassis},

		{`SELECT id::text FROM inv.devices
		    WHERE tenant_id=$1 AND deleted_at IS NULL AND base_mac = $2::macaddr LIMIT 1`,
			n.GetRemoteMac(), confMAC},

		// MAC сусіда може збігтися не з шасі, а з конкретним портом.
		{`SELECT i.device_id::text FROM inv.interfaces i
		    WHERE i.tenant_id=$1 AND i.mac = $2::macaddr LIMIT 1`,
			n.GetRemoteMac(), confMAC},

		{`SELECT id::text FROM inv.devices
		    WHERE tenant_id=$1 AND deleted_at IS NULL AND host(address) = $2 LIMIT 1`,
			n.GetRemoteMgmtIp(), confMgmtIP},

		{`SELECT id::text FROM inv.devices
		    WHERE tenant_id=$1 AND deleted_at IS NULL
		      AND (system_name = $2 OR lower(name) = lower($2)) LIMIT 1`,
			n.GetRemoteSystemName(), confSysName},
	}

	for _, p := range probes {
		if p.arg == "" {
			continue
		}
		var id string
		err := tx.QueryRow(ctx, p.query, tenantID, p.arg).Scan(&id)
		if errors.Is(err, pgx.ErrNoRows) {
			continue
		}
		if err != nil {
			// Некоректний MAC/IP від пристрою не має валити весь звіт.
			continue
		}
		return &id, p.conf, nil
	}
	return nil, 0, nil
}

// resolveRemoteInterface: port-id у LLDP буває чим завгодно —
// ім'ям порту, описом, ifIndex або MAC. Пробуємо все.
func (s *Store) resolveRemoteInterface(ctx context.Context, tx pgx.Tx, tenantID, deviceID, portID, portDescr string) (any, error) {
	candidates := []string{portID, portDescr}

	for _, c := range candidates {
		if strings.TrimSpace(c) == "" {
			continue
		}
		var id string

		err := tx.QueryRow(ctx, `
			SELECT id::text FROM inv.interfaces
			WHERE tenant_id=$1 AND device_id=$2
			  AND (lower(name) = lower($3) OR lower(alias) = lower($3))
			LIMIT 1
		`, tenantID, deviceID, c).Scan(&id)
		if err == nil {
			return id, nil
		}
		if !errors.Is(err, pgx.ErrNoRows) {
			return nil, err
		}

		if idx, convErr := strconv.ParseInt(strings.TrimSpace(c), 10, 64); convErr == nil {
			err = tx.QueryRow(ctx, `
				SELECT id::text FROM inv.interfaces
				WHERE tenant_id=$1 AND device_id=$2 AND if_index=$3 LIMIT 1
			`, tenantID, deviceID, idx).Scan(&id)
			if err == nil {
				return id, nil
			}
			if !errors.Is(err, pgx.ErrNoRows) {
				return nil, err
			}
		}
	}
	return nil, nil
}

type linkSpec struct {
	aDevice    string
	aInterface any
	bDevice    string
	bInterface any
	proto      string
	confidence int
}

// upsertLink створює або оновлює лінк, нормалізуючи пару.
//
// Пошук іде за тим самим виразом LEAST/GREATEST, що й унікальний
// індекс links_pair_uniq, тому A→B і B→A знаходять один рядок.
// Лінк із is_pinned не перезаписується: підтверджене людиною має
// пріоритет над автовиявленням, інакше кожен запуск discovery
// затирав би ручні правки.
func (s *Store) upsertLink(ctx context.Context, tx pgx.Tx, tenantID string, spec linkSpec) (created bool, err error) {
	const zero = "00000000-0000-0000-0000-000000000000"

	var (
		linkID   string
		isPinned bool
	)
	err = tx.QueryRow(ctx, `
		SELECT id::text, is_pinned FROM topo.links
		WHERE tenant_id = $1
		  AND LEAST(a_device_id, b_device_id) = LEAST($2::uuid, $3::uuid)
		  AND GREATEST(a_device_id, b_device_id) = GREATEST($2::uuid, $3::uuid)
		  AND LEAST(COALESCE(a_interface_id, $6::uuid), COALESCE(b_interface_id, $6::uuid))
		      = LEAST(COALESCE($4::uuid, $6::uuid), COALESCE($5::uuid, $6::uuid))
		  AND GREATEST(COALESCE(a_interface_id, $6::uuid), COALESCE(b_interface_id, $6::uuid))
		      = GREATEST(COALESCE($4::uuid, $6::uuid), COALESCE($5::uuid, $6::uuid))
	`, tenantID, spec.aDevice, spec.bDevice, spec.aInterface, spec.bInterface, zero).
		Scan(&linkID, &isPinned)

	switch {
	case err == nil:
		if isPinned {
			// Лише позначаємо, що лінк живий.
			_, err = tx.Exec(ctx, `
				UPDATE topo.links SET last_seen_at = now() WHERE id = $1
			`, linkID)
			return false, err
		}
		_, err = tx.Exec(ctx, `
			UPDATE topo.links
			SET last_seen_at = now(),
			    confidence = GREATEST(confidence, $2),
			    discovered_by = $3::topo.discovery_proto
			WHERE id = $1
		`, linkID, spec.confidence, spec.proto)
		return false, err

	case errors.Is(err, pgx.ErrNoRows):
		// Швидкість каналу беремо з порту: вона стане знаменником
		// для util_pct і, зрештою, швидкістю анімації на мапі.
		_, err = tx.Exec(ctx, `
			INSERT INTO topo.links
				(tenant_id, a_device_id, a_interface_id, b_device_id, b_interface_id,
				 kind, capacity_bps, discovered_by, confidence, status, last_seen_at)
			VALUES ($1,$2,$3,$4,$5,'physical',
			        (SELECT speed_bps FROM inv.interfaces WHERE id = $3::uuid),
			        $6::topo.discovery_proto, $7, 'unknown', now())
			ON CONFLICT DO NOTHING
		`, tenantID, spec.aDevice, spec.aInterface, spec.bDevice, spec.bInterface,
			spec.proto, spec.confidence)
		return err == nil, err

	default:
		return false, err
	}
}

func protoName(p npv1.DiscoveryProto) string {
	switch p {
	case npv1.DiscoveryProto_DISCOVERY_PROTO_LLDP:
		return "lldp"
	case npv1.DiscoveryProto_DISCOVERY_PROTO_CDP:
		return "cdp"
	case npv1.DiscoveryProto_DISCOVERY_PROTO_ARP:
		return "arp"
	case npv1.DiscoveryProto_DISCOVERY_PROTO_FDB:
		return "fdb"
	case npv1.DiscoveryProto_DISCOVERY_PROTO_STP:
		return "stp"
	case npv1.DiscoveryProto_DISCOVERY_PROTO_ROUTING:
		return "routing"
	case npv1.DiscoveryProto_DISCOVERY_PROTO_SNMP_TOPO:
		return "snmp_topo"
	default:
		return "manual"
	}
}

func statusOr(s, def string) string {
	if s == "" {
		return def
	}
	return s
}

func nullInt(v int) any {
	if v == 0 {
		return nil
	}
	return v
}

func nullInt64(v int64) any {
	if v == 0 {
		return nil
	}
	return v
}