node/message/outbound_msg_builder.go

// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.

package message

import (
	"net/netip"
	"time"

	compression "github.com/luxfi/compress"
	"github.com/luxfi/ids"
	"github.com/luxfi/net/endpoints"
	"github.com/luxfi/node/proto/p2p"
)

var _ OutboundMsgBuilder = (*outMsgBuilder)(nil)

// OutboundMsgBuilder builds outbound messages. Outbound messages are returned
// with a reference count of 1. Once the reference count hits 0, the message
// bytes should no longer be accessed.
type OutboundMsgBuilder interface {
	Handshake(
		networkID uint32,
		myTime uint64,
		ip netip.AddrPort,
		client string,
		major uint32,
		minor uint32,
		patch uint32,
		ipSigningTime uint64,
		ipNodeIDSig []byte,
		ipBLSSig []byte,
		trackedNets []ids.ID,
		supportedLPs []uint32,
		objectedLPs []uint32,
		knownPeersFilter []byte,
		knownPeersSalt []byte,
		requestAllNetIPs bool,
		ipMLDSASig []byte,
	) (OutboundMessage, error)

	GetPeerList(
		knownPeersFilter []byte,
		knownPeersSalt []byte,
		requestAllNetIPs bool,
	) (OutboundMessage, error)

	PeerList(
		peers []*endpoints.ClaimedIPPort,
		bypassThrottling bool,
	) (OutboundMessage, error)

	Ping(
		primaryUptime uint32,
	) (OutboundMessage, error)

	Pong() (OutboundMessage, error)

	GetStateSummaryFrontier(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
	) (OutboundMessage, error)

	StateSummaryFrontier(
		chainID ids.ID,
		requestID uint32,
		summary []byte,
	) (OutboundMessage, error)

	GetAcceptedStateSummary(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
		heights []uint64,
	) (OutboundMessage, error)

	AcceptedStateSummary(
		chainID ids.ID,
		requestID uint32,
		summaryIDs []ids.ID,
	) (OutboundMessage, error)

	GetAcceptedFrontier(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
	) (OutboundMessage, error)

	AcceptedFrontier(
		chainID ids.ID,
		requestID uint32,
		containerID ids.ID,
	) (OutboundMessage, error)

	GetAccepted(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
		containerIDs []ids.ID,
	) (OutboundMessage, error)

	Accepted(
		chainID ids.ID,
		requestID uint32,
		containerIDs []ids.ID,
	) (OutboundMessage, error)

	GetAncestors(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
		containerID ids.ID,
		engineType p2p.EngineType,
	) (OutboundMessage, error)

	Ancestors(
		chainID ids.ID,
		requestID uint32,
		containers [][]byte,
	) (OutboundMessage, error)

	Get(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
		containerID ids.ID,
	) (OutboundMessage, error)

	Put(
		chainID ids.ID,
		requestID uint32,
		container []byte,
	) (OutboundMessage, error)

	PushQuery(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
		container []byte,
		requestedHeight uint64,
	) (OutboundMessage, error)

	PullQuery(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
		containerID ids.ID,
		requestedHeight uint64,
	) (OutboundMessage, error)

	Chits(
		chainID ids.ID,
		requestID uint32,
		preferredID ids.ID,
		preferredIDAtHeight ids.ID,
		acceptedID ids.ID,
		acceptedHeight uint64,
	) (OutboundMessage, error)

	Request(
		chainID ids.ID,
		requestID uint32,
		deadline time.Duration,
		msg []byte,
	) (OutboundMessage, error)

	Response(
		chainID ids.ID,
		requestID uint32,
		msg []byte,
	) (OutboundMessage, error)

	Error(
		chainID ids.ID,
		requestID uint32,
		errorCode int32,
		errorMessage string,
	) (OutboundMessage, error)

	Gossip(
		chainID ids.ID,
		msg []byte,
	) (OutboundMessage, error)

	BFTMessage(
		msg *p2p.BFT,
	) (OutboundMessage, error)
}

type outMsgBuilder struct {
	compressionType compression.Type

	builder *msgBuilder
}

// Use "message.NewCreator" to import this function
// since we do not expose "msgBuilder" yet
func newOutboundBuilder(compressionType compression.Type, builder *msgBuilder) OutboundMsgBuilder {
	return &outMsgBuilder{
		compressionType: compressionType,
		builder:         builder,
	}
}

// ... ping/pong omitted for brevity in prompt, only replacing targeted section ...
// Wait, replace_file_content replaces a chunk. I need to target the Interface definitions and Implementation.
// Interface is lines 160-188.
// Implementation is lines 663-731.
// I should use 2 ReplaceFile calls or one if contiguous? They are NOT contiguous.
// I can use `multi_replace_file_content`? Or `replace_file_content` twice.
// I'll do Interface first.
// Oh wait, `replace_file_content` only allows ONE chunk.
// I'll use separate calls or `multi_replace`.
// I'll use `multi_replace_file_content` for `outbound_msg_builder.go`.

func (b *outMsgBuilder) Ping(
	primaryUptime uint32,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Ping{
				Ping: &p2p.Ping{
					Uptime: primaryUptime,
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) Pong() (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Pong{
				Pong: &p2p.Pong{},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) Handshake(
	networkID uint32,
	myTime uint64,
	ip netip.AddrPort,
	client string,
	major uint32,
	minor uint32,
	patch uint32,
	ipSigningTime uint64,
	ipNodeIDSig []byte,
	ipBLSSig []byte,
	trackedNets []ids.ID,
	supportedLPs []uint32,
	objectedLPs []uint32,
	knownPeersFilter []byte,
	knownPeersSalt []byte,
	requestAllNetIPs bool,
	ipMLDSASig []byte,
) (OutboundMessage, error) {
	subsubchainIDBytes := make([][]byte, len(trackedNets))
	encodeIDs(trackedNets, subsubchainIDBytes)
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Handshake{
				Handshake: &p2p.Handshake{
					NetworkId:     networkID,
					MyTime:        myTime,
					IpAddr:        ip.Addr().AsSlice(),
					IpPort:        uint32(ip.Port()),
					IpSigningTime: ipSigningTime,
					IpNodeIdSig:   ipNodeIDSig,
					TrackedNets:   subsubchainIDBytes,
					Client: &p2p.Client{
						Name:  client,
						Major: major,
						Minor: minor,
						Patch: patch,
					},
					SupportedLps: supportedLPs,
					ObjectedLps:  objectedLPs,
					KnownPeers: &p2p.BloomFilter{
						Filter: knownPeersFilter,
						Salt:   knownPeersSalt,
					},
					IpBlsSig:   ipBLSSig,
					AllChains:  requestAllNetIPs,
					IpMldsaSig: ipMLDSASig,
				},
			},
		},
		compression.TypeNone,
		true,
	)
}

func (b *outMsgBuilder) GetPeerList(
	knownPeersFilter []byte,
	knownPeersSalt []byte,
	requestAllNetIPs bool,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_GetPeerList{
				GetPeerList: &p2p.GetPeerList{
					KnownPeers: &p2p.BloomFilter{
						Filter: knownPeersFilter,
						Salt:   knownPeersSalt,
					},
					AllChains: requestAllNetIPs,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) PeerList(peers []*endpoints.ClaimedIPPort, bypassThrottling bool) (OutboundMessage, error) {
	claimIPPorts := make([]*p2p.ClaimedIpPort, len(peers))
	for i, p := range peers {
		claimIPPorts[i] = &p2p.ClaimedIpPort{
			X509Certificate: p.Cert.Raw,
			IpAddr:          p.AddrPort.Addr().AsSlice(),
			IpPort:          uint32(p.AddrPort.Port()),
			Timestamp:       p.Timestamp,
			Signature:       p.Signature,
			TxId:            ids.Empty[:],
		}
	}
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_PeerList_{
				PeerList_: &p2p.PeerList{
					ClaimedIpPorts: claimIPPorts,
				},
			},
		},
		b.compressionType,
		bypassThrottling,
	)
}

func (b *outMsgBuilder) GetStateSummaryFrontier(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_GetStateSummaryFrontier{
				GetStateSummaryFrontier: &p2p.GetStateSummaryFrontier{
					ChainId:   chainID[:],
					RequestId: requestID,
					Deadline:  uint64(deadline),
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) StateSummaryFrontier(
	chainID ids.ID,
	requestID uint32,
	summary []byte,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_StateSummaryFrontier_{
				StateSummaryFrontier_: &p2p.StateSummaryFrontier{
					ChainId:   chainID[:],
					RequestId: requestID,
					Summary:   summary,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) GetAcceptedStateSummary(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
	heights []uint64,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_GetAcceptedStateSummary{
				GetAcceptedStateSummary: &p2p.GetAcceptedStateSummary{
					ChainId:   chainID[:],
					RequestId: requestID,
					Deadline:  uint64(deadline),
					Heights:   heights,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) AcceptedStateSummary(
	chainID ids.ID,
	requestID uint32,
	summaryIDs []ids.ID,
) (OutboundMessage, error) {
	summaryIDBytes := make([][]byte, len(summaryIDs))
	encodeIDs(summaryIDs, summaryIDBytes)
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_AcceptedStateSummary_{
				AcceptedStateSummary_: &p2p.AcceptedStateSummary{
					ChainId:    chainID[:],
					RequestId:  requestID,
					SummaryIds: summaryIDBytes,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) GetAcceptedFrontier(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_GetAcceptedFrontier{
				GetAcceptedFrontier: &p2p.GetAcceptedFrontier{
					ChainId:   chainID[:],
					RequestId: requestID,
					Deadline:  uint64(deadline),
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) AcceptedFrontier(
	chainID ids.ID,
	requestID uint32,
	containerID ids.ID,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_AcceptedFrontier_{
				AcceptedFrontier_: &p2p.AcceptedFrontier{
					ChainId:     chainID[:],
					RequestId:   requestID,
					ContainerId: containerID[:],
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) GetAccepted(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
	containerIDs []ids.ID,
) (OutboundMessage, error) {
	containerIDBytes := make([][]byte, len(containerIDs))
	encodeIDs(containerIDs, containerIDBytes)
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_GetAccepted{
				GetAccepted: &p2p.GetAccepted{
					ChainId:      chainID[:],
					RequestId:    requestID,
					Deadline:     uint64(deadline),
					ContainerIds: containerIDBytes,
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) Accepted(
	chainID ids.ID,
	requestID uint32,
	containerIDs []ids.ID,
) (OutboundMessage, error) {
	containerIDBytes := make([][]byte, len(containerIDs))
	encodeIDs(containerIDs, containerIDBytes)
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Accepted_{
				Accepted_: &p2p.Accepted{
					ChainId:      chainID[:],
					RequestId:    requestID,
					ContainerIds: containerIDBytes,
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) GetAncestors(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
	containerID ids.ID,
	engineType p2p.EngineType,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_GetAncestors{
				GetAncestors: &p2p.GetAncestors{
					ChainId:     chainID[:],
					RequestId:   requestID,
					Deadline:    uint64(deadline),
					ContainerId: containerID[:],
					EngineType:  engineType,
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) Ancestors(
	chainID ids.ID,
	requestID uint32,
	containers [][]byte,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Ancestors_{
				Ancestors_: &p2p.Ancestors{
					ChainId:    chainID[:],
					RequestId:  requestID,
					Containers: containers,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) Get(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
	containerID ids.ID,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Get{
				Get: &p2p.Get{
					ChainId:     chainID[:],
					RequestId:   requestID,
					Deadline:    uint64(deadline),
					ContainerId: containerID[:],
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) Put(
	chainID ids.ID,
	requestID uint32,
	container []byte,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Put{
				Put: &p2p.Put{
					ChainId:   chainID[:],
					RequestId: requestID,
					Container: container,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) PushQuery(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
	container []byte,
	requestedHeight uint64,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_PushQuery{
				PushQuery: &p2p.PushQuery{
					ChainId:         chainID[:],
					RequestId:       requestID,
					Deadline:        uint64(deadline),
					Container:       container,
					RequestedHeight: requestedHeight,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) PullQuery(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
	containerID ids.ID,
	requestedHeight uint64,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_PullQuery{
				PullQuery: &p2p.PullQuery{
					ChainId:         chainID[:],
					RequestId:       requestID,
					Deadline:        uint64(deadline),
					ContainerId:     containerID[:],
					RequestedHeight: requestedHeight,
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) Chits(
	chainID ids.ID,
	requestID uint32,
	preferredID ids.ID,
	preferredIDAtHeight ids.ID,
	acceptedID ids.ID,
	acceptedHeight uint64,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Chits{
				Chits: &p2p.Chits{
					ChainId:             chainID[:],
					RequestId:           requestID,
					PreferredId:         preferredID[:],
					PreferredIdAtHeight: preferredIDAtHeight[:],
					AcceptedId:          acceptedID[:],
					AcceptedHeight:      acceptedHeight,
				},
			},
		},
		compression.TypeNone,
		false,
	)
}

func (b *outMsgBuilder) Request(
	chainID ids.ID,
	requestID uint32,
	deadline time.Duration,
	msg []byte,
) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Request{
				Request: &p2p.Request{
					ChainId:   chainID[:],
					RequestId: requestID,
					Deadline:  uint64(deadline),
					AppBytes:  msg,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) Response(chainID ids.ID, requestID uint32, msg []byte) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Response{
				Response: &p2p.Response{
					ChainId:   chainID[:],
					RequestId: requestID,
					AppBytes:  msg,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) Error(chainID ids.ID, requestID uint32, errorCode int32, errorMessage string) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Error{
				Error: &p2p.Error{
					ChainId:      chainID[:],
					RequestId:    requestID,
					ErrorCode:    errorCode,
					ErrorMessage: errorMessage,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) Gossip(chainID ids.ID, msg []byte) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: &p2p.Message_Gossip{
				Gossip: &p2p.Gossip{
					ChainId:  chainID[:],
					AppBytes: msg,
				},
			},
		},
		b.compressionType,
		false,
	)
}

func (b *outMsgBuilder) BFTMessage(msg *p2p.BFT) (OutboundMessage, error) {
	return b.builder.createOutbound(
		&p2p.Message{
			Message: newMessageBFT(msg),
		},
		b.compressionType,
		false,
	)
}