node/indexer/indexer.go

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

package indexer

import (
	"fmt"
	"io"
	"sync"

	"github.com/zap-proto/zip"

	"github.com/luxfi/database"
	"github.com/luxfi/database/prefixdb"
	"github.com/luxfi/ids"
	"github.com/luxfi/log"
	"github.com/luxfi/node/chains"
	nodeconsensus "github.com/luxfi/node/consensus"
	"github.com/luxfi/node/server/http"
	"github.com/luxfi/node/utils/wrappers"
	"github.com/luxfi/runtime"
	"github.com/luxfi/timer/mockable"
	"github.com/luxfi/vm"
)

const (
	indexNamePrefix         = "index-"
	txPrefix                = 0x01
	vtxPrefix               = 0x02
	blockPrefix             = 0x03
	isIncompletePrefix      = 0x04
	previouslyIndexedPrefix = 0x05
)

var (
	_ Indexer = (*indexer)(nil)

	hasRunKey = []byte{0x07}
)

// Config for an indexer
type Config struct {
	DB                   database.Database
	Log                  log.Logger
	IndexingEnabled      bool
	AllowIncompleteIndex bool
	BlockAcceptorGroup   nodeconsensus.AcceptorGroup
	TxAcceptorGroup      nodeconsensus.AcceptorGroup
	VertexAcceptorGroup  nodeconsensus.AcceptorGroup
	APIServer            server.PathAdder
	ShutdownF            func()
}

// Indexer causes accepted containers for a given chain
// to be indexed by their ID and by the order in which
// they were accepted by this node.
// Indexer is threadsafe.
type Indexer interface {
	chains.Registrant
	// Close will do nothing and return nil after the first call
	io.Closer
}

// NewIndexer returns a new Indexer and registers a new endpoint on the given API server.
func NewIndexer(config Config) (Indexer, error) {
	indexer := &indexer{
		clock:                &mockable.Clock{},
		log:                  config.Log,
		db:                   config.DB,
		allowIncompleteIndex: config.AllowIncompleteIndex,
		indexingEnabled:      config.IndexingEnabled,
		blockAcceptorGroup:   config.BlockAcceptorGroup,
		txAcceptorGroup:      config.TxAcceptorGroup,
		vertexAcceptorGroup:  config.VertexAcceptorGroup,
		txIndices:            map[ids.ID]*index{},
		vtxIndices:           map[ids.ID]*index{},
		blockIndices:         map[ids.ID]*index{},
		pathAdder:            config.APIServer,
		shutdownF:            config.ShutdownF,
	}

	hasRun, err := indexer.hasRun()
	if err != nil {
		return nil, err
	}
	indexer.hasRunBefore = hasRun
	return indexer, indexer.markHasRun()
}

type indexer struct {
	clock  *mockable.Clock
	lock   sync.RWMutex
	log    log.Logger
	db     database.Database
	closed bool

	// Called in a goroutine on shutdown
	shutdownF func()

	// true if this is not the first run using this database
	hasRunBefore bool

	// Used to add API endpoint for new indices
	pathAdder server.PathAdder

	// If true, allow running in such a way that could allow the creation
	// of an index which could be missing accepted containers.
	allowIncompleteIndex bool

	// If false, don't create index for a chain when RegisterChain is called
	indexingEnabled bool

	// The operations served for each index, held so shutdown can stop the
	// listeners they run on.
	ops []*zip.App

	// Chain ID --> index of blocks of that chain (if applicable)
	blockIndices map[ids.ID]*index
	// Chain ID --> index of vertices of that chain (if applicable)
	vtxIndices map[ids.ID]*index
	// Chain ID --> index of txs of that chain (if applicable)
	txIndices map[ids.ID]*index

	// Notifies of newly accepted blocks
	blockAcceptorGroup nodeconsensus.AcceptorGroup
	// Notifies of newly accepted transactions
	txAcceptorGroup nodeconsensus.AcceptorGroup
	// Notifies of newly accepted vertices
	vertexAcceptorGroup nodeconsensus.AcceptorGroup
}

// RegisterChain registers a chain for indexing
func (i *indexer) RegisterChain(chainName string, rt *runtime.Runtime, vm vm.VM) {
	i.lock.Lock()
	defer i.lock.Unlock()

	// Extract chain ID from runtime
	chainID := rt.ChainID

	if i.closed {
		i.log.Debug("not registering chain to indexer",
			log.String("reason", "indexer is closed"),
			log.String("chainName", chainName),
		)
		return
	} else if rt.NetworkID != 1 && rt.NetworkID != 2 {
		// Only index chains on mainnet (1) or testnet (2) - skip custom networks
		i.log.Debug("not registering chain to indexer",
			log.String("reason", "not on mainnet or testnet"),
			log.String("chainName", chainName),
		)
		return
	}
	if i.blockIndices[chainID] != nil || i.txIndices[chainID] != nil || i.vtxIndices[chainID] != nil {
		i.log.Warn("chain is already being indexed",
			log.Stringer("chainID", chainID),
		)
		return
	}

	// If the index is incomplete, make sure that's OK. Otherwise, cause node to die.
	isIncomplete, err := i.isIncomplete(chainID)
	if err != nil {
		i.log.Error("couldn't get whether chain is incomplete",
			log.String("chainName", chainName),
			log.Err(err),
		)
		if err := i.close(); err != nil {
			i.log.Error("failed to close indexer",
				log.Err(err),
			)
		}
		return
	}

	// See if this chain was indexed in a previous run
	previouslyIndexed, err := i.previouslyIndexed(chainID)
	if err != nil {
		i.log.Error("couldn't get whether chain was previously indexed",
			log.String("chainName", chainName),
			log.Err(err),
		)
		if err := i.close(); err != nil {
			i.log.Error("failed to close indexer",
				log.Err(err),
			)
		}
		return
	}

	if !i.indexingEnabled { // Indexing is disabled
		if previouslyIndexed && !i.allowIncompleteIndex {
			// We indexed this chain in a previous run but not in this run.
			// This would create an incomplete index, which is not allowed, so exit.
			i.log.Error("running would cause index to become incomplete but incomplete indices are disabled",
				log.String("chainName", chainName),
			)
			if err := i.close(); err != nil {
				i.log.Error("failed to close indexer",
					log.Err(err),
				)
			}
			return
		}

		// Creating an incomplete index is allowed. Mark index as incomplete.
		err := i.markIncomplete(chainID)
		if err == nil {
			return
		}
		i.log.Error("couldn't mark chain as incomplete",
			log.String("chainName", chainName),
			log.Err(err),
		)
		if err := i.close(); err != nil {
			i.log.Error("failed to close indexer",
				log.Err(err),
			)
		}
		return
	}

	if !i.allowIncompleteIndex && isIncomplete && (previouslyIndexed || i.hasRunBefore) {
		i.log.Error("index is incomplete but incomplete indices are disabled. Shutting down",
			log.String("chainName", chainName),
		)
		if err := i.close(); err != nil {
			i.log.Error("failed to close indexer",
				log.Err(err),
			)
		}
		return
	}

	// Mark that in this run, this chain was indexed
	if err := i.markPreviouslyIndexed(chainID); err != nil {
		i.log.Error("couldn't mark chain as indexed",
			log.String("chainName", chainName),
			log.Err(err),
		)
		if err := i.close(); err != nil {
			i.log.Error("failed to close indexer",
				log.Err(err),
			)
		}
		return
	}

	index, err := i.registerChainHelper(chainID, blockPrefix, chainName, "block", i.blockAcceptorGroup)
	if err != nil {
		i.log.Error("failed to create index",
			log.String("chainName", chainName),
			log.String("endpoint", "block"),
			log.Err(err),
			log.String("debug", "closing indexer due to block index creation failure"),
		)
		if err := i.close(); err != nil {
			i.log.Error("failed to close indexer",
				log.Err(err),
			)
		}
		return
	}
	i.blockIndices[chainID] = index

	// Currently supporting block-based VM indexing via core.VM interface
	// DAG-based VM indexing is fully supported in the consensus package
	vmType := fmt.Sprintf("%T", vm)
	i.log.Debug("RegisterChain completed",
		log.String("chainName", chainName),
		log.String("vmType", vmType),
		log.Stringer("chainID", chainID),
	)
}

func (i *indexer) registerChainHelper(
	chainID ids.ID,
	prefixEnd byte,
	name, endpoint string,
	acceptorGroup nodeconsensus.AcceptorGroup,
) (*index, error) {
	prefix := make([]byte, ids.IDLen+wrappers.ByteLen)
	copy(prefix, chainID[:])
	prefix[ids.IDLen] = prefixEnd
	indexDB := prefixdb.New(prefix, i.db)
	index, err := newIndex(indexDB, i.log, i.clock)
	if err != nil {
		// Don't close indexDB as it would close the underlying database
		// The prefixdb doesn't hold resources that need explicit cleanup
		return nil, err
	}

	// Register index to learn about new accepted vertices
	if err := acceptorGroup.RegisterAcceptor(chainID, fmt.Sprintf("%s%s", indexNamePrefix, chainID), index, true); err != nil {
		_ = index.Close()
		return nil, err
	}

	// Serve this index's operations beneath its own endpoint, so
	// /v1/index/<chain>/<what>/ops is where they answer.
	app, handler, err := (&service{index: index}).mount(i.log, name+"-"+endpoint)
	if err != nil {
		_ = index.Close()
		return nil, err
	}
	i.ops = append(i.ops, app)
	if err := i.pathAdder.AddRoute(handler, "index/"+name, "/"+endpoint+"/ops"); err != nil {
		_ = index.Close()
		return nil, err
	}
	return index, nil
}

// Close this indexer. Stops indexing all chains.
// Closes [i.db]. Assumes Close is only called after
// the node is done making decisions.
// Calling Close after it has been called does nothing.
func (i *indexer) Close() error {
	i.lock.Lock()
	defer i.lock.Unlock()

	return i.close()
}

func (i *indexer) close() error {
	if i.closed {
		return nil
	}
	i.closed = true

	errs := &wrappers.Errs{}
	for chainID, txIndex := range i.txIndices {
		errs.Add(
			txIndex.Close(),
			i.txAcceptorGroup.DeregisterAcceptor(chainID, fmt.Sprintf("%s%s", indexNamePrefix, chainID)),
		)
	}
	for chainID, vtxIndex := range i.vtxIndices {
		errs.Add(
			vtxIndex.Close(),
			i.vertexAcceptorGroup.DeregisterAcceptor(chainID, fmt.Sprintf("%s%s", indexNamePrefix, chainID)),
		)
	}
	for chainID, blockIndex := range i.blockIndices {
		errs.Add(
			blockIndex.Close(),
			i.blockAcceptorGroup.DeregisterAcceptor(chainID, fmt.Sprintf("%s%s", indexNamePrefix, chainID)),
		)
	}
	for _, app := range i.ops {
		errs.Add(app.Shutdown())
	}
	errs.Add(i.db.Close())

	go i.shutdownF()
	return errs.Err
}

func (i *indexer) markIncomplete(chainID ids.ID) error {
	key := make([]byte, ids.IDLen+wrappers.ByteLen)
	copy(key, chainID[:])
	key[ids.IDLen] = isIncompletePrefix
	return i.db.Put(key, nil)
}

// Returns true if this chain is incomplete
func (i *indexer) isIncomplete(chainID ids.ID) (bool, error) {
	key := make([]byte, ids.IDLen+wrappers.ByteLen)
	copy(key, chainID[:])
	key[ids.IDLen] = isIncompletePrefix
	return i.db.Has(key)
}

func (i *indexer) markPreviouslyIndexed(chainID ids.ID) error {
	key := make([]byte, ids.IDLen+wrappers.ByteLen)
	copy(key, chainID[:])
	key[ids.IDLen] = previouslyIndexedPrefix
	return i.db.Put(key, nil)
}

// Returns true if this chain is incomplete
func (i *indexer) previouslyIndexed(chainID ids.ID) (bool, error) {
	key := make([]byte, ids.IDLen+wrappers.ByteLen)
	copy(key, chainID[:])
	key[ids.IDLen] = previouslyIndexedPrefix
	return i.db.Has(key)
}

// Mark that the node has run at least once
func (i *indexer) markHasRun() error {
	return i.db.Put(hasRunKey, nil)
}

// Returns true if the node has run before
func (i *indexer) hasRun() (bool, error) {
	return i.db.Has(hasRunKey)
}