routewatch/internal/ristypes/ris.go

// Package ristypes defines the data structures for RIS Live BGP messages and
// announcements.
package ristypes

import (
	"encoding/json"
	"time"
)

// Message types and the session state the handlers act on, as the feed writes
// them in a message's type and state fields.
const (
	// MessageTypeUpdate is a BGP UPDATE: announcements and withdrawals.
	MessageTypeUpdate = "UPDATE"

	// MessageTypeState is a change in the state of the BGP session between
	// the collector and the peer.
	MessageTypeState = "STATE"

	// StateDown is the state of a STATE message whose session has ended: the
	// collector no longer has any of the peer's routes, and sends no
	// withdrawals for them.
	StateDown = "down"
)

// ASPath is a BGP AS path as the feed writes it: an array of AS numbers in
// which an AS set is an array of its members, such as [1, 2, [3, 4]].
type ASPath struct {
	// Sequence is the path's AS numbers in order. A set of one member that
	// ends the path counts as that AS and is its last element. The members of
	// a set anywhere else are flattened into it.
	Sequence []int

	// Set holds the members of a set of several members that ends the path.
	// Any of them may be the origin, and they are not neighbours of each
	// other or of the AS before them.
	Set []int
}

// Origin returns the AS that originated the route: the last AS of the
// sequence, or 0 (none) when the path ends in a set of several members or is
// empty.
func (p *ASPath) Origin() int {
	if len(p.Set) > 0 || len(p.Sequence) == 0 {
		return 0
	}

	return p.Sequence[len(p.Sequence)-1]
}

// UnmarshalJSON implements the json.Unmarshaler interface for ASPath.
func (p *ASPath) UnmarshalJSON(data []byte) error {
	// Most paths have no AS set and decode as a plain array of integers.
	var sequence []int

	err := json.Unmarshal(data, &sequence)
	if err == nil {
		*p = ASPath{Sequence: sequence}

		return nil
	}

	var raw []any

	err = json.Unmarshal(data, &raw)
	if err != nil {
		return err
	}

	path := ASPath{Sequence: make([]int, 0, len(raw))}

	for i, item := range raw {
		switch v := item.(type) {
		case float64:
			path.Sequence = append(path.Sequence, int(v))
		case []any:
			set := make([]int, 0, len(v))

			for _, member := range v {
				if num, ok := member.(float64); ok {
					set = append(set, int(num))
				}
			}

			if i == len(raw)-1 && len(set) > 1 {
				path.Set = set
			} else {
				path.Sequence = append(path.Sequence, set...)
			}
		}
	}

	*p = path

	return nil
}

// RISLiveMessage represents the outer wrapper message from the RIPE RIS Live stream.
// Each message contains a Type field indicating the message type and a Data field
// containing the actual BGP message payload.
type RISLiveMessage struct {
	Type string     `json:"type"`
	Data RISMessage `json:"data"`
}

// RISMessage represents a BGP update message from the RIPE RIS Live stream.
// It contains metadata about the BGP session (peer, ASN, host) along with
// the actual BGP update data including AS path, communities, announcements,
// and withdrawals.
//
//nolint:tagliatelle // field names are the RIS Live feed's
type RISMessage struct {
	Type            string    `json:"type"`
	Timestamp       float64   `json:"timestamp"`
	ParsedTimestamp time.Time `json:"-"` // Parsed from Timestamp field
	Peer            string    `json:"peer"`
	PeerASN         string    `json:"peer_asn"`
	ID              string    `json:"id"`
	Host            string    `json:"host"`
	RRC             string    `json:"rrc,omitempty"`
	MrtTime         float64   `json:"mrt_time,omitempty"`
	SocketTime      float64   `json:"socket_time,omitempty"`
	Path            ASPath    `json:"path,omitzero"`
	// Community and Raw are present in the feed but read by no handler.
	// They are the largest fields on a message that lives in up to four
	// handler queues, so json:"-" keeps them out of the decoded message
	// to save queue memory. Do not decode them without a consumer.
	Community     [][]int           `json:"-"`
	Origin        string            `json:"origin,omitempty"`
	MED           *int              `json:"med,omitempty"`
	LocalPref     *int              `json:"local_pref,omitempty"`
	Announcements []RISAnnouncement `json:"announcements,omitempty"`
	Withdrawals   []string          `json:"withdrawals,omitempty"`
	Raw           string            `json:"-"`
	// State is set on STATE messages: the session's new state, such as StateDown.
	State string `json:"state,omitempty"`
}

// RISAnnouncement represents a BGP route announcement within a RIS message.
// It contains the next hop IP address and the list of prefixes being announced
// via that next hop.
//
//nolint:tagliatelle // field names are the RIS Live feed's
type RISAnnouncement struct {
	NextHop  string   `json:"next_hop"`
	Prefixes []string `json:"prefixes"`
}