routewatch/internal/streamer/streamer.go

// Package streamer implements an HTTP client that connects to the RIPE RIS Live
// streaming API, parses BGP UPDATE messages from the JSON stream, and
// dispatches them to registered handlers.
package streamer

import (
	"bufio"
	"compress/gzip"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"math/rand"
	"net/http"
	"sync"
	"sync/atomic"
	"time"

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

// countingReader wraps an io.Reader and counts bytes read
type countingReader struct {
	reader io.Reader
	count  atomic.Int64
}

// Read implements io.Reader and counts bytes
func (c *countingReader) Read(p []byte) (int, error) {
	n, err := c.reader.Read(p)
	c.count.Add(int64(n))

	return n, err
}

// Count returns the total bytes read
func (c *countingReader) Count() int64 {
	return c.count.Load()
}

// Configuration constants for the RIS Live streamer.
const (
	risLiveURL = "https://ris-live.ripe.net/v1/stream/?format=json&" +
		"client=https%3A%2F%2Fgit.eeqj.de%2Fsneak%2Froutewatch"
	metricsWindowSize = 60 // seconds for rolling average
	metricsUpdateRate = time.Second
	minBackoffDelay   = 5 * time.Second
	maxBackoffDelay   = 320 * time.Second
	// Reconnect when no line arrives for this long
	streamIdleTimeout  = 60 * time.Second
	metricsLogInterval = 10 * time.Second
	bytesPerKB         = 1024
	bytesPerMB         = 1024 * 1024
	// Well above an extended BGP message (RFC 8654) in hex plus its prefixes
	maxLineSize           = bytesPerMB
	maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers

	// Backpressure constants
	backpressureThreshold = 0.5 // Start dropping at 50% queue utilization
	backpressureSlope     = 2.0 // Slope for linear drop probability increase
)

// messageTypeNotification is the type of a BGP NOTIFICATION message, which ends
// the session with the peer.
const messageTypeNotification = "NOTIFICATION"

var (
	errAlreadyRunning   = errors.New("streamer already running")
	errUnexpectedStatus = errors.New("unexpected status code")
	// errStreamEnded is returned by stream when RIS Live ends the response.
	errStreamEnded = errors.New("RIS Live ended the stream")
	// errStreamIdle is returned by stream when no line arrived for the idle
	// timeout.
	errStreamIdle = errors.New("no line received")
)

// MessageHandler defines the interface for processing RIS messages.
// Implementations must specify which message types they want to receive,
// how to process messages, and their desired queue capacity.
type MessageHandler interface {
	// WantsMessage returns true if this handler wants to process messages of
	// the given type.
	WantsMessage(messageType string) bool

	// HandleMessage processes a RIS message. This method is called from a dedicated
	// goroutine for each handler, so implementations do not need to be goroutine-safe
	// with respect to other handlers.
	HandleMessage(msg *ristypes.RISMessage)

	// QueueCapacity returns the desired queue capacity for this handler.
	// Handlers that process quickly can have larger queues to buffer bursts.
	// When the queue fills up, messages will be dropped according to the
	// backpressure algorithm, except STATE messages, which are held until
	// there is room.
	QueueCapacity() int
}

// RawMessageHandler is a function type for processing raw JSON lines from the stream.
// It receives the unmodified JSON line as a string before any parsing occurs.
type RawMessageHandler func(line string)

// handlerMetrics tracks performance metrics for a handler
type handlerMetrics struct {
	processedCount     uint64        // Total messages processed
	droppedCount       uint64        // Total messages dropped
	totalTime          time.Duration // Total processing time (for average calculation)
	minTime            time.Duration // Minimum processing time
	maxTime            time.Duration // Maximum processing time
	queueHighWaterMark int           // Maximum queue length seen
	mu                 sync.Mutex    // Protects the metrics
}

// handlerInfo wraps a handler with its queue and metrics
type handlerInfo struct {
	handler MessageHandler
	queue   chan *ristypes.RISMessage
	metrics handlerMetrics
	// heldState holds, in feed order, the STATE messages that found queue
	// full (see queueMessage). Only the read loop uses it, and then Stop.
	heldState []*ristypes.RISMessage
}

// Streamer manages a connection to the RIPE RIS Live streaming API for receiving
// real-time BGP UPDATE messages. It handles automatic reconnection with exponential
// backoff, dispatches messages to registered handlers via per-handler queues, and
// implements backpressure to prevent queue overflow during high traffic periods.
type Streamer struct {
	logger *logger.Logger
	client *http.Client
	url    string
	// Drop the connection when no line arrives for this long
	idleTimeout time.Duration
	// First wait before reconnecting; doubles up to maxBackoffDelay
	minBackoff   time.Duration
	handlers     []*handlerInfo
	rawHandler   RawMessageHandler
	mu           sync.RWMutex
	cancel       context.CancelFunc
	running      bool
	workers      sync.WaitGroup // the handler workers; Stop waits for them
	metrics      *metrics.Tracker
	totalDropped atomic.Uint64   // Total dropped messages across all handlers
	random       *rand.Rand      // Random number generator for backpressure drops
	bgpPeers     map[string]bool // Track active BGP peers by peer IP
	bgpPeersMu   sync.RWMutex    // Protects bgpPeers map
}

// New creates a new Streamer instance configured to connect to the RIS Live API.
// The logger is used for structured logging of connection events and errors.
// The metrics tracker is used to record message counts, bytes received, and
// connection status.
func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
	return &Streamer{
		logger:      logger,
		url:         risLiveURL,
		idleTimeout: streamIdleTimeout,
		minBackoff:  minBackoffDelay,
		client: &http.Client{
			Timeout: 0, // No timeout for streaming
			Transport: &http.Transport{
				// Disable automatic gzip decompression so we can measure wire bytes
				DisableCompression: true,
			},
		},
		handlers: make([]*handlerInfo, 0),
		metrics:  metrics,
		//nolint:gosec // Non-cryptographic randomness is fine for backpressure
		random:   rand.New(rand.NewSource(time.Now().UnixNano())),
		bgpPeers: make(map[string]bool),
	}
}

// RegisterHandler adds a MessageHandler to receive parsed RIS messages.
// Each handler gets its own dedicated queue and worker goroutine for processing.
// If the streamer is already running, the handler's worker is started immediately.
func (s *Streamer) RegisterHandler(handler MessageHandler) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Create handler info with its own queue based on capacity
	info := &handlerInfo{
		handler: handler,
		queue:   make(chan *ristypes.RISMessage, handler.QueueCapacity()),
		metrics: handlerMetrics{
			// Initialize to max so first value sets the floor
			minTime: time.Duration(math.MaxInt64),
		},
	}

	s.handlers = append(s.handlers, info)

	// If we're already running, start a worker for this handler
	if s.running {
		s.workers.Add(1)

		go s.runHandlerWorker(info)
	}
}

// SetURL makes the streamer read from url instead of RIS Live. Call it before
// Start. Tests use it to read from a local server.
func (s *Streamer) SetURL(url string) {
	s.url = url
}

// RegisterRawHandler sets a callback to receive raw JSON lines from the stream
// before they are parsed. Only one raw handler can be registered at a time;
// subsequent calls will replace the previous handler.
func (s *Streamer) RegisterRawHandler(handler RawMessageHandler) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.rawHandler = handler
}

// Start begins streaming BGP updates from the RIS Live API in a background goroutine.
// It starts worker goroutines for each registered handler and manages automatic
// reconnection with exponential backoff on connection failures.
// Returns an error if the streamer is already running. It streams until Stop:
// the end of ctx does not stop it.
func (s *Streamer) Start(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.running {
		return errAlreadyRunning
	}

	ctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
	s.cancel = cancel
	s.running = true

	// Start workers for each handler
	for _, info := range s.handlers {
		s.workers.Add(1)

		go s.runHandlerWorker(info)
	}

	go func() {
		s.streamWithReconnect(ctx)
		s.mu.Lock()
		s.running = false
		s.mu.Unlock()
	}()

	return nil
}

// Stop ends the stream and stops the handler workers. It returns once each
// worker has handed its handler every message left on its queue and then the
// STATE messages held for it (see queueMessage), or once ctx ends, and then
// logs how many messages were not handed over. This method is safe to call
// multiple times.
func (s *Streamer) Stop(ctx context.Context) {
	s.mu.Lock()
	if s.cancel == nil {
		// Not started, or already stopped: closing the queues again would panic.
		s.mu.Unlock()

		return
	}

	// Once the stream's context is cancelled the read loop queues and holds
	// nothing more (see dispatchMessage).
	s.cancel()
	s.cancel = nil

	notHandedOver := 0

	for _, info := range s.handlers {
		// The held messages came after everything on the queue. Each waits
		// for the worker, which does not take s.mu, to make room, until ctx
		// ends.
		for len(info.heldState) > 0 && ctx.Err() == nil {
			select {
			case info.queue <- info.heldState[0]:
				info.heldState = info.heldState[1:]
			case <-ctx.Done():
			}
		}

		notHandedOver += len(info.heldState)
		info.heldState = nil

		// The worker returns once it has emptied the closed queue.
		close(info.queue)
	}

	handlers := s.handlers
	s.running = false
	s.mu.Unlock()
	s.metrics.SetConnected(false)

	workersDone := make(chan struct{})

	go func() {
		s.workers.Wait()
		close(workersDone)
	}()

	select {
	case <-workersDone:
	case <-ctx.Done():
		for _, info := range handlers {
			notHandedOver += len(info.queue)
		}
	}

	if notHandedOver > 0 {
		s.logger.Error("Stop deadline passed before the handlers took every "+
			"message", "messages", notHandedOver)
	}
}

// IsRunning reports whether the streamer is currently connected and processing
// messages. This is safe to call concurrently from multiple goroutines.
func (s *Streamer) IsRunning() bool {
	s.mu.RLock()
	defer s.mu.RUnlock()

	return s.running
}

// GetMetrics returns the current streaming metrics including message counts,
// bytes received, and throughput rates. The returned struct is a snapshot
// of the current state and is safe to use without synchronization.
func (s *Streamer) GetMetrics() metrics.StreamMetrics {
	return s.metrics.GetStreamMetrics()
}

// GetMetricsTracker returns the underlying metrics.Tracker instance for direct access
// to metrics recording and retrieval functionality.
func (s *Streamer) GetMetricsTracker() *metrics.Tracker {
	return s.metrics
}

// HandlerStats contains performance metrics for a single message handler.
// It includes queue utilization, message counts, and processing time statistics.
type HandlerStats struct {
	Name               string
	QueueLength        int
	QueueCapacity      int
	QueueHighWaterMark int
	ProcessedCount     uint64
	DroppedCount       uint64
	AvgProcessTime     time.Duration
	MinProcessTime     time.Duration
	MaxProcessTime     time.Duration
}

// GetHandlerStats returns a snapshot of performance statistics for all registered
// handlers. The returned slice contains one HandlerStats entry per handler with
// current queue depth, processed/dropped counts, and processing time statistics.
func (s *Streamer) GetHandlerStats() []HandlerStats {
	s.mu.RLock()
	defer s.mu.RUnlock()

	stats := make([]HandlerStats, 0, len(s.handlers))

	for _, info := range s.handlers {
		info.metrics.mu.Lock()

		hs := HandlerStats{
			Name:               fmt.Sprintf("%T", info.handler),
			QueueLength:        len(info.queue),
			QueueCapacity:      cap(info.queue),
			QueueHighWaterMark: info.metrics.queueHighWaterMark,
			ProcessedCount:     info.metrics.processedCount,
			DroppedCount:       info.metrics.droppedCount,
			MinProcessTime:     info.metrics.minTime,
			MaxProcessTime:     info.metrics.maxTime,
		}

		// Calculate average time
		if info.metrics.processedCount > 0 {
			processedCount := info.metrics.processedCount

			const maxInt64 = 1<<63 - 1

			if processedCount > maxInt64 {
				processedCount = maxInt64
			}

			hs.AvgProcessTime = info.metrics.totalTime /
				time.Duration(processedCount)
		}

		info.metrics.mu.Unlock()

		stats = append(stats, hs)
	}

	return stats
}

// GetDroppedMessages returns the total number of messages dropped across all
// handlers due to queue overflow or backpressure. This counter is monotonically
// increasing and is safe to call concurrently.
func (s *Streamer) GetDroppedMessages() uint64 {
	return s.totalDropped.Load()
}

// runHandlerWorker processes messages for a specific handler
func (s *Streamer) runHandlerWorker(info *handlerInfo) {
	defer s.workers.Done()

	for msg := range info.queue {
		start := time.Now()

		info.handler.HandleMessage(msg)

		elapsed := time.Since(start)

		// Update metrics
		info.metrics.mu.Lock()
		info.metrics.processedCount++
		info.metrics.totalTime += elapsed

		// Update min time
		if elapsed < info.metrics.minTime {
			info.metrics.minTime = elapsed
		}

		// Update max time
		if elapsed > info.metrics.maxTime {
			info.metrics.maxTime = elapsed
		}
		info.metrics.mu.Unlock()
	}
}

// logMetrics logs the current streaming statistics
func (s *Streamer) logMetrics() {
	metrics := s.metrics.GetStreamMetrics()
	uptime := time.Since(metrics.ConnectedSince)

	const bitsPerMegabit = 1000000

	totalDropped := s.totalDropped.Load()

	s.logger.Info(
		"Stream statistics",
		"uptime",
		uptime,
		"total_messages",
		metrics.TotalMessages,
		"wire_bytes",
		metrics.TotalWireBytes,
		"wire_mb",
		fmt.Sprintf("%.2f", float64(metrics.TotalWireBytes)/bytesPerMB),
		"wire_mbps",
		fmt.Sprintf("%.2f", metrics.WireBitsPerSec/bitsPerMegabit),
		"decompressed_bytes",
		metrics.TotalBytes,
		"decompressed_mb",
		fmt.Sprintf("%.2f", float64(metrics.TotalBytes)/bytesPerMB),
		"messages_per_sec",
		fmt.Sprintf("%.2f", metrics.MessagesPerSec),
		"total_dropped",
		totalDropped,
	)

	// Log per-handler statistics
	s.mu.RLock()

	for i, info := range s.handlers {
		info.metrics.mu.Lock()
		if info.metrics.processedCount > 0 {
			// Safe conversion: processedCount is bounded by maxInt64
			processedCount := info.metrics.processedCount

			const maxInt64 = 1<<63 - 1

			if processedCount > maxInt64 {
				processedCount = maxInt64
			}

			avgTime := info.metrics.totalTime / time.Duration(processedCount)
			s.logger.Info(
				"Handler statistics",
				"handler", fmt.Sprintf("%T", info.handler),
				"index", i,
				"queue_len", len(info.queue),
				"queue_cap", cap(info.queue),
				"processed", info.metrics.processedCount,
				"dropped", info.metrics.droppedCount,
				"avg_time", avgTime,
				"min_time", info.metrics.minTime,
				"max_time", info.metrics.maxTime,
			)
		}
		info.metrics.mu.Unlock()
	}

	s.mu.RUnlock()
}

// updateMetrics updates the metrics counters and rates
func (s *Streamer) updateMetrics(messageBytes int) {
	s.metrics.RecordMessage(int64(messageBytes))
}

// streamWithReconnect handles streaming with automatic reconnection and
// exponential backoff
func (s *Streamer) streamWithReconnect(ctx context.Context) {
	backoffDelay := s.minBackoff
	consecutiveFailures := 0

	for {
		select {
		case <-ctx.Done():
			s.logger.Info("Stream context cancelled, stopping reconnection attempts")

			return
		default:
		}

		// Attempt to stream
		startTime := time.Now()
		err := s.stream(ctx)
		streamDuration := time.Since(startTime)

		// Only Stop ends the streamer. A stream that ends any other way,
		// with or without an error, is a disconnect and is reconnected.
		if ctx.Err() != nil {
			return
		}

		// Log the error
		s.logger.Error("Stream disconnected",
			"error", err,
			"consecutive_failures", consecutiveFailures+1,
			"stream_duration", streamDuration)
		s.metrics.SetConnected(false)

		// A stream that lasted more than 30 seconds resets the backoff. That
		// includes a stream dropped after streamIdleTimeout with no line, so
		// a silent feed is retried after the first backoff delay every time.
		if streamDuration > 30*time.Second {
			s.logger.Info("Resetting backoff delay due to successful connection",
				"stream_duration", streamDuration)

			backoffDelay = s.minBackoff
			consecutiveFailures = 0
		} else {
			// Increment consecutive failures
			consecutiveFailures++
		}

		// Wait with exponential backoff
		s.logger.Info("Waiting before reconnection attempt",
			"delay_seconds", backoffDelay.Seconds(),
			"consecutive_failures", consecutiveFailures)

		select {
		case <-ctx.Done():
			return
		case <-time.After(backoffDelay):
			// Double the backoff delay for next time, up to max
			backoffDelay *= 2
			if backoffDelay > maxBackoffDelay {
				backoffDelay = maxBackoffDelay
			}
		}
	}
}

func (s *Streamer) stream(ctx context.Context) error {
	// connCtx is scoped to this single connection: cancelling it when stream
	// returns stops the ticker goroutines below, so a reconnect does not leak
	// them. Without this they would live until the streamer's lifetime context
	// is cancelled, leaking two per reconnect.
	connCtx, connCancel := context.WithCancel(ctx)
	defer connCancel()

	// The idle timer drops the connection, by cancelling connCtx, when no
	// line arrives for s.idleTimeout. Each line read starts it over. It
	// also covers the wait for the server to answer.
	idleTimer := time.AfterFunc(s.idleTimeout, connCancel)
	defer idleTimer.Stop()

	req, err := http.NewRequestWithContext(connCtx, http.MethodGet, s.url, nil)
	if err != nil {
		return fmt.Errorf("failed to create request: %w", err)
	}

	// Explicitly request gzip compression
	req.Header.Set("Accept-Encoding", "gzip")

	resp, err := s.client.Do(req)
	if err != nil {
		return fmt.Errorf("failed to connect to RIS Live: %w", err)
	}
	defer func() {
		err := resp.Body.Close()
		if err != nil {
			s.logger.Error("Failed to close response body", "error", err)
		}
	}()

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%w: %d", errUnexpectedStatus, resp.StatusCode)
	}

	// Wrap body with counting reader to track actual wire bytes
	wireCounter := &countingReader{reader: resp.Body}

	// Check if response is gzip-compressed and decompress if needed
	var reader io.Reader = wireCounter
	if resp.Header.Get("Content-Encoding") == "gzip" {
		gzReader, err := gzip.NewReader(wireCounter)
		if err != nil {
			return fmt.Errorf("failed to create gzip reader: %w", err)
		}
		defer func() { _ = gzReader.Close() }()

		reader = gzReader

		s.logger.Info("Connected to RIS Live stream", "compression", "gzip")
	} else {
		s.logger.Info("Connected to RIS Live stream", "compression", "none")
	}

	s.metrics.SetConnected(true)

	// Start metrics logging goroutine
	metricsTicker := time.NewTicker(metricsLogInterval)
	defer metricsTicker.Stop()

	go func() {
		for {
			select {
			case <-metricsTicker.C:
				s.logMetrics()
			case <-connCtx.Done():
				return
			}
		}
	}()

	// Wire byte update ticker - update metrics with actual wire bytes periodically
	go s.recordWireBytes(connCtx, wireCounter)

	return s.readStream(ctx, connCtx, reader, idleTimer)
}

// recordWireBytes adds the bytes wireCounter has counted to the metrics once a
// second until ctx is cancelled.
func (s *Streamer) recordWireBytes(
	ctx context.Context, wireCounter *countingReader,
) {
	wireUpdateTicker := time.NewTicker(time.Second)
	defer wireUpdateTicker.Stop()

	var lastWireBytes int64

	for {
		select {
		case <-wireUpdateTicker.C:
			currentBytes := wireCounter.Count()

			delta := currentBytes - lastWireBytes
			if delta > 0 {
				s.metrics.RecordWireBytes(delta)

				lastWireBytes = currentBytes
			}
		case <-ctx.Done():
			return
		}
	}
}

// readStream hands each line of the stream to processLine until the stream
// ends, a line fails, or ctx is cancelled. Each line restarts idleTimer, which
// cancels connCtx when it fires.
func (s *Streamer) readStream(
	ctx, connCtx context.Context, reader io.Reader, idleTimer *time.Timer,
) error {
	scanner := bufio.NewScanner(reader)
	scanner.Buffer(nil, maxLineSize)

	for scanner.Scan() {
		idleTimer.Reset(s.idleTimeout)

		select {
		case <-ctx.Done():
			s.logger.Info("Stream stopped by context")

			return ctx.Err()
		default:
		}

		line := scanner.Bytes()
		if len(line) == 0 {
			continue
		}

		err := s.processLine(ctx, line)
		if err != nil {
			return err
		}
	}

	if ctx.Err() != nil {
		return ctx.Err()
	}

	if connCtx.Err() != nil {
		return fmt.Errorf("%w for %s", errStreamIdle, s.idleTimeout)
	}

	err := scanner.Err()
	if err != nil {
		return fmt.Errorf("scanner error: %w", err)
	}

	return errStreamEnded
}

// processLine passes one line of the stream to the raw handler, parses it,
// updates the peer set (updatePeers), and dispatches an UPDATE or STATE
// message to the handlers. An error ends the connection.
func (s *Streamer) processLine(ctx context.Context, line []byte) error {
	// Update metrics with decompressed message size
	s.updateMetrics(len(line))

	// Call raw handler if registered
	s.mu.RLock()
	rawHandler := s.rawHandler
	s.mu.RUnlock()

	if rawHandler != nil {
		// Call raw handler synchronously to preserve order
		rawHandler(string(line))
	}

	// Parse the message first
	var wrapper ristypes.RISLiveMessage

	err := json.Unmarshal(line, &wrapper)
	if err != nil {
		// Log the error and return to trigger reconnection
		s.logger.Error("Failed to parse JSON",
			"error", err,
			"line", string(line),
			"line_length", len(line))

		return fmt.Errorf("JSON parse error: %w", err)
	}

	// Check if it's a ris_message wrapper
	if wrapper.Type != "ris_message" {
		s.logger.Error("Unexpected wrapper type",
			"type", wrapper.Type,
			"line", string(line),
		)

		return nil
	}

	// Get the actual message
	msg := wrapper.Data

	// Parse the timestamp
	msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()

	peerCount := s.updatePeers(&msg)

	// Process based on message type
	switch msg.Type {
	case ristypes.MessageTypeUpdate:
		// Process BGP UPDATE messages
		// Will be dispatched to handlers
	case "RIS_PEER_STATE":
		// RIS peer state messages - silently ignore
		return nil
	case "KEEPALIVE":
		// BGP keepalive messages - silently process
		return nil
	case "OPEN":
		s.logger.Info("BGP session opened",
			"peer", msg.Peer,
			"peer_asn", msg.PeerASN,
			"total_peers", peerCount,
		)

		return nil
	case messageTypeNotification:
		s.logger.Warn("BGP notification",
			"peer", msg.Peer,
			"peer_asn", msg.PeerASN,
			"total_peers", peerCount,
		)

		return nil
	case ristypes.MessageTypeState:
		// The BGP session with the peer changed state, for example went
		// down. Dispatched to the handlers that want it.
	default:
		s.logger.Warn("Unknown message type, skipping",
			"type", msg.Type,
		)

		return nil
	}

	return s.dispatchMessage(ctx, &msg)
}

// updatePeers adds the peer msg came from to the peer set, or removes it if
// msg ends the peer's session: a NOTIFICATION, or a STATE message whose state
// is down. It records the size of the set as the BGP peer count and returns
// it.
func (s *Streamer) updatePeers(msg *ristypes.RISMessage) int {
	sessionEnded := msg.Type == messageTypeNotification ||
		(msg.Type == ristypes.MessageTypeState && msg.State == ristypes.StateDown)

	s.bgpPeersMu.Lock()

	if sessionEnded {
		delete(s.bgpPeers, msg.Peer)
	} else {
		s.bgpPeers[msg.Peer] = true
	}

	peerCount := len(s.bgpPeers)
	s.bgpPeersMu.Unlock()
	s.metrics.SetBGPPeerCount(peerCount)

	return peerCount
}

// dispatchMessage queues msg for each handler that wants it, unless
// queueMessage drops it for that handler.
func (s *Streamer) dispatchMessage(
	ctx context.Context, msg *ristypes.RISMessage,
) error {
	// Stop cancels ctx and closes the queues under the write lock, so if ctx
	// is cancelled here, under the read lock, the queues are closed and must
	// not be sent to.
	s.mu.RLock()

	if ctx.Err() != nil {
		s.mu.RUnlock()

		return ctx.Err()
	}

	for _, info := range s.handlers {
		if !info.handler.WantsMessage(msg.Type) {
			continue
		}

		if !s.queueMessage(info, msg) {
			atomic.AddUint64(&info.metrics.droppedCount, 1)
			s.totalDropped.Add(1)

			continue
		}

		// Update high water mark if needed
		queueLen := len(info.queue)
		info.metrics.mu.Lock()
		if queueLen > info.metrics.queueHighWaterMark {
			info.metrics.queueHighWaterMark = queueLen
		}
		info.metrics.mu.Unlock()
	}

	s.mu.RUnlock()

	return nil
}

// queueMessage puts msg on the handler's queue and reports false if it dropped
// msg instead. It never waits for room, so a full queue does not hold up the
// read loop.
//
// A message is dropped while the queue is filling up (shouldDropForBackpressure)
// or full, except a STATE message: one whose state is down is the only notice
// that all of a peer's routes are gone. A STATE message skips the random drop,
// and one that finds the queue full is held in info.heldState. Held messages
// go into the queue, oldest first, before anything else for the handler, as
// soon as there is room, or when Stop is called; while any is still held, the
// handler's other messages are dropped, so none of them overtakes a held down.
//
// The held list stays small: a STATE message is a few hundred bytes and the
// feed sends about ten a second, so a queue full for a minute holds about 600.
func (s *Streamer) queueMessage(
	info *handlerInfo, msg *ristypes.RISMessage,
) bool {
	for len(info.heldState) > 0 && trySend(info.queue, info.heldState[0]) {
		info.heldState = info.heldState[1:]
	}

	if msg.Type == ristypes.MessageTypeState {
		if len(info.heldState) > 0 || !trySend(info.queue, msg) {
			info.heldState = append(info.heldState, msg)
		}

		return true
	}

	if len(info.heldState) > 0 || s.shouldDropForBackpressure(info) {
		return false
	}

	return trySend(info.queue, msg)
}

// trySend puts msg on queue if there is room and reports whether it did.
func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool {
	select {
	case queue <- msg:
		return true
	default:
		return false
	}
}

// shouldDropForBackpressure determines if a message should be dropped based on
// queue utilization
func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
	// Calculate queue utilization
	queueLen := len(info.queue)
	queueCap := cap(info.queue)
	utilization := float64(queueLen) / float64(queueCap)

	// No drops below threshold
	if utilization < backpressureThreshold {
		return false
	}

	// Calculate drop probability (0.0 at threshold, 1.0 at 100% full)
	dropProbability := (utilization - backpressureThreshold) * backpressureSlope
	if dropProbability > 1.0 {
		dropProbability = 1.0
	}

	// Random drop based on probability
	return s.random.Float64() < dropProbability
}