routewatch/internal/routewatch/peeringhandler.go

package routewatch

import (
	"encoding/json"
	"sync"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/database"
	"git.eeqj.de/sneak/routewatch/internal/logger"
	"git.eeqj.de/sneak/routewatch/internal/ristypes"
)

const (
	// peeringHandlerQueueSize is the buffer capacity for the peering handler's
	// message queue, about 4 seconds of feed at peak. The streamer drops
	// rather than blocks when a queue is full, so this bounds memory.
	peeringHandlerQueueSize = 20000

	// minPathLengthForPeering specifies the minimum number of ASNs required
	// in a BGP AS path to extract peering relationships. A path with fewer
	// than 2 ASNs cannot contain any peering information.
	minPathLengthForPeering = 2

	// peeringProcessInterval controls how frequently the handler processes
	// accumulated AS paths and extracts peering relationships to store
	// in the database.
	peeringProcessInterval = 30 * time.Second

	// maxTrackedPaths bounds how many distinct AS paths are held in memory
	// between processing runs. Once the map is full, further new paths are
	// dropped and counted until the next run empties it.
	maxTrackedPaths = 500000
)

// PeeringHandler processes BGP UPDATE messages to extract and track
// AS peering relationships. It accumulates AS paths in memory and
// periodically processes them to extract unique peering pairs, which
// are then stored in the database. The handler implements the Handler
// interface for integration with the message processing pipeline.
type PeeringHandler struct {
	db     database.Store
	logger *logger.Logger

	// In-memory AS path tracking
	mu           sync.Mutex
	asPaths      map[string]time.Time // key is JSON-encoded AS path
	droppedPaths int                  // paths dropped because the map was full

	stopCh chan struct{}
}

// NewPeeringHandler creates and initializes a new PeeringHandler with the
// provided database store and logger. It starts one background goroutine that
// periodically processes accumulated AS paths into peering records. The
// handler begins processing immediately upon creation.
func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler {
	h := &PeeringHandler{
		db:      db,
		logger:  logger,
		asPaths: make(map[string]time.Time),
		stopCh:  make(chan struct{}),
	}

	// Start the periodic processing goroutine
	go h.processLoop()

	return h
}

// WantsMessage reports whether the handler should receive messages of the
// given type. PeeringHandler only processes UPDATE messages, as these contain
// the AS path information needed to extract peering relationships.
func (h *PeeringHandler) WantsMessage(messageType string) bool {
	// We only care about UPDATE messages that have AS paths
	return messageType == "UPDATE"
}

// QueueCapacity returns the buffer size for the handler's message queue.
// This value is used by the message dispatcher to allocate the channel
// buffer when registering the handler.
func (h *PeeringHandler) QueueCapacity() int {
	return peeringHandlerQueueSize
}

// HandleMessage processes a BGP UPDATE message by storing its AS path
// in memory for later batch processing. Messages with AS paths shorter
// than minPathLengthForPeering are ignored as they cannot contain valid
// peering information.
func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
	// Skip if no AS path or only one AS
	if len(msg.Path) < minPathLengthForPeering {
		return
	}

	timestamp := msg.ParsedTimestamp

	// Encode AS path as JSON for use as map key
	pathJSON, err := json.Marshal(msg.Path)
	if err != nil {
		h.logger.Error("Failed to encode AS path", "error", err)

		return
	}
	key := string(pathJSON)

	h.mu.Lock()
	if _, exists := h.asPaths[key]; exists {
		// Already tracked: refresh its timestamp.
		h.asPaths[key] = timestamp
	} else if len(h.asPaths) >= maxTrackedPaths {
		// Map is full; drop this new path and count it.
		h.droppedPaths++
	} else {
		h.asPaths[key] = timestamp
	}
	h.mu.Unlock()
}

// processLoop runs periodically to process AS paths into peerings
func (h *PeeringHandler) processLoop() {
	ticker := time.NewTicker(peeringProcessInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			h.processPeerings()
		case <-h.stopCh:
			// Final processing
			h.processPeerings()

			return
		}
	}
}

// ProcessPeeringsNow triggers immediate processing of all accumulated AS
// paths into peering records. This bypasses the normal periodic processing
// schedule and is primarily intended for testing purposes.
func (h *PeeringHandler) ProcessPeeringsNow() {
	h.processPeerings()
}

// processPeerings extracts peerings from AS paths and writes to database
func (h *PeeringHandler) processPeerings() {
	// Take the accumulated paths and replace the map with a fresh empty one
	// under the lock. Each path is processed exactly once and the memory is
	// released, so the map never grows past a single interval's traffic.
	h.mu.Lock()
	paths := h.asPaths
	h.asPaths = make(map[string]time.Time)
	dropped := h.droppedPaths
	h.mu.Unlock()

	if len(paths) == 0 {
		return
	}

	// Extract unique peerings from AS paths
	type peeringKey struct {
		low, high int
	}
	peerings := make(map[peeringKey]time.Time)

	for pathJSON, timestamp := range paths {
		var path []int
		if err := json.Unmarshal([]byte(pathJSON), &path); err != nil {
			h.logger.Error("Failed to decode AS path", "error", err)

			continue
		}

		// Extract peerings from path
		for i := range len(path) - 1 {
			asn1 := path[i]
			asn2 := path[i+1]

			// Skip invalid ASNs
			if asn1 <= 0 || asn2 <= 0 || asn1 == asn2 {
				continue
			}

			// Normalize: lower AS number first
			low, high := asn1, asn2
			if low > high {
				low, high = high, low
			}

			key := peeringKey{low: low, high: high}
			// Update timestamp if this is newer
			if existing, ok := peerings[key]; !ok || timestamp.After(existing) {
				peerings[key] = timestamp
			}
		}
	}

	// Record peerings in database
	start := time.Now()
	successCount := 0
	for key, ts := range peerings {
		err := h.db.RecordPeering(key.low, key.high, ts)
		if err != nil {
			h.logger.Error("Failed to record peering",
				"as_a", key.low,
				"as_b", key.high,
				"error", err,
			)
		} else {
			successCount++
		}
	}

	h.logger.Info("Processed AS peerings",
		"paths", len(paths),
		"unique_peerings", len(peerings),
		"success", successCount,
		"dropped_paths", dropped,
		"duration", time.Since(start),
	)
}

// Stop gracefully shuts down the handler by signaling the background
// goroutine to stop and performing a final synchronous processing of
// any remaining AS paths. This ensures no peering data is lost during
// shutdown.
func (h *PeeringHandler) Stop() {
	close(h.stopCh)
	// Process any remaining peerings synchronously
	h.processPeerings()
}