warp/chainipc.go

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

package warp

import (
	"fmt"
	"maps"
	"path/filepath"
	"slices"

	"github.com/luxfi/ids"
	log "github.com/luxfi/log"
)

const (
	// DefaultBaseURL can be used as a reasonable default value for the base URL
	DefaultBaseURL = "/tmp"

	ipcIdentifierPrefix    = "ipc"
	ipcConsensusIdentifier = "consensus"
	ipcDecisionsIdentifier = "decisions"
)

type ipcContext struct {
	log       log.Logger
	networkID uint32
	path      string
}

// ChainIPCs maintains IPCs for a set of chains
type ChainIPCs struct {
	ipcContext
	chains              map[ids.ID]*EventSockets
	blockAcceptorGroup  AcceptorGroup
	txAcceptorGroup     AcceptorGroup
	vertexAcceptorGroup AcceptorGroup
}

// NewChainIPCs creates a new *ChainIPCs that writes consensus and decision
// events to IPC sockets
func NewChainIPCs(
	log log.Logger,
	path string,
	networkID uint32,
	blockAcceptorGroup AcceptorGroup,
	txAcceptorGroup AcceptorGroup,
	vertexAcceptorGroup AcceptorGroup,
	defaultChainIDs []ids.ID,
) (*ChainIPCs, error) {
	cipcs := &ChainIPCs{
		ipcContext: ipcContext{
			log:       log,
			networkID: networkID,
			path:      path,
		},
		chains:              make(map[ids.ID]*EventSockets),
		blockAcceptorGroup:  blockAcceptorGroup,
		txAcceptorGroup:     txAcceptorGroup,
		vertexAcceptorGroup: vertexAcceptorGroup,
	}
	for _, chainID := range defaultChainIDs {
		if _, err := cipcs.Publish(chainID); err != nil {
			return nil, err
		}
	}
	return cipcs, nil
}

// Publish creates a set of eventSockets for the given chainID
func (cipcs *ChainIPCs) Publish(chainID ids.ID) (*EventSockets, error) {
	if es, ok := cipcs.chains[chainID]; ok {
		cipcs.log.Info("returning existing event sockets",
			log.Stringer("blockchainID", chainID),
		)
		return es, nil
	}

	es, err := newEventSockets(
		cipcs.ipcContext,
		chainID,
		cipcs.blockAcceptorGroup,
		cipcs.txAcceptorGroup,
		cipcs.vertexAcceptorGroup,
	)
	if err != nil {
		cipcs.log.Error("can't create ipcs",
			log.Reflect("error", err),
		)
		return nil, err
	}

	cipcs.chains[chainID] = es
	cipcs.log.Info("created IPC sockets",
		log.Stringer("blockchainID", chainID),
		log.UserString("consensusURL", es.ConsensusURL()),
		log.UserString("decisionsURL", es.DecisionsURL()),
	)
	return es, nil
}

// Unpublish stops the eventSocket for the given chain if it exists. It returns
// whether or not the socket existed and errors when trying to close it
func (cipcs *ChainIPCs) Unpublish(chainID ids.ID) (bool, error) {
	chainIPCs, ok := cipcs.chains[chainID]
	if !ok {
		return false, nil
	}
	delete(cipcs.chains, chainID)
	return true, chainIPCs.stop()
}

// GetPublishedBlockchains returns the chains that are currently being published
func (cipcs *ChainIPCs) GetPublishedBlockchains() []ids.ID {
	return slices.Collect(maps.Keys(cipcs.chains))
}

func (cipcs *ChainIPCs) Shutdown() error {
	cipcs.log.Info("shutting down chain IPCs")

	errs := firstErr{}
	for _, ch := range cipcs.chains {
		errs.add(ch.stop())
	}
	return errs.err
}

func ipcURL(ctx ipcContext, chainID ids.ID, eventType string) string {
	return filepath.Join(ctx.path, fmt.Sprintf("%d-%s-%s", ctx.networkID, chainID.String(), eventType))
}