routewatch/internal/routewatch/peeringhandler.go

package routewatch

import (
	"context"
	"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: each path, JSON-encoded, with the earliest
	// and the latest time an update carried it.
	mu           sync.Mutex
	asPaths      map[string]database.SeenTimes
	droppedPaths int // paths dropped because the map was full
	// writing is how many paths processPeerings took from asPaths and has
	// not finished writing the peerings of.
	writing int

	// stopCh hands Stop's context to the processing goroutine. It holds one,
	// so Stop does not wait for a write in progress to hand it over.
	stopCh chan context.Context
	done   chan struct{} // closed when the processing goroutine returns
}

// 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]database.SeenTimes),
		stopCh:  make(chan context.Context, 1),
		done:    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 == ristypes.MessageTypeUpdate
}

// 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) {
	// Peerings are between neighbours in the path's sequence only: the
	// members of an AS set that ends the path are not neighbours of each
	// other or of the AS before them.
	sequence := msg.Path.Sequence

	// Skip if no AS path or only one AS
	if len(sequence) < minPathLengthForPeering {
		return
	}

	timestamp := msg.ParsedTimestamp

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

		return
	}

	key := string(pathJSON)

	h.mu.Lock()
	defer h.mu.Unlock()

	seen, exists := h.asPaths[key]
	if !exists && len(h.asPaths) >= maxTrackedPaths {
		// Map is full; drop this new path and count it.
		h.droppedPaths++

		return
	}

	seen.Add(timestamp)
	h.asPaths[key] = seen
}

// 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()
}

// Stop stops the processing goroutine, which writes the peerings of the AS
// paths the handler holds before it returns, and waits for it until ctx ends.
// Once ctx has ended the processing goroutine skips that last write, and a
// write still running is lost when the process exits, so Stop logs how many
// paths' peerings are not written.
func (h *PeeringHandler) Stop(ctx context.Context) {
	h.stopCh <- ctx

	select {
	case <-h.done:
	case <-ctx.Done():
	}

	h.mu.Lock()
	unwritten := h.writing + len(h.asPaths)
	h.mu.Unlock()

	if unwritten > 0 {
		h.logger.Error("Stop deadline passed before the peering handler wrote "+
			"its peerings", "paths", unwritten)
	}
}

// processLoop runs periodically to process AS paths into peerings, and once
// more on Stop.
func (h *PeeringHandler) processLoop() {
	defer close(h.done)

	ticker := time.NewTicker(peeringProcessInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			h.processPeerings()
		case ctx := <-h.stopCh:
			// Final processing, unless the stop deadline has passed
			if ctx.Err() == nil {
				h.processPeerings()
			}

			return
		}
	}
}

// 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]database.SeenTimes)
	h.writing = len(paths)
	dropped := h.droppedPaths
	h.mu.Unlock()

	if len(paths) == 0 {
		return
	}

	// Extract unique peerings from AS paths
	type peeringKey struct {
		low, high int
	}

	// The earliest and the latest time an update carried each peering. Several
	// paths, each carried at several times, can hold a peering, so a new
	// peering is stored first seen at the earliest of them and last seen at the
	// latest.
	peerings := make(map[peeringKey]database.SeenTimes)

	for pathJSON, pathSeen := range paths {
		var path []int

		err := json.Unmarshal([]byte(pathJSON), &path)
		if 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}

			seen := peerings[key]
			seen.Add(pathSeen.First)
			seen.Add(pathSeen.Last)
			peerings[key] = seen
		}
	}

	// Record peerings in database
	start := time.Now()
	successCount := 0

	for key, seen := range peerings {
		err := h.db.RecordPeering(key.low, key.high, seen)
		if err != nil {
			h.logger.Error("Failed to record peering",
				"as_a", key.low,
				"as_b", key.high,
				"error", err,
			)
		} else {
			successCount++
		}
	}

	h.mu.Lock()
	h.writing = 0
	h.mu.Unlock()

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