components.go

package block

import (
	"context"
	"errors"
	"fmt"

	"github.com/celestiaorg/go-header"
	"github.com/rs/zerolog"

	"github.com/evstack/ev-node/block/internal/cache"
	"github.com/evstack/ev-node/block/internal/common"
	da "github.com/evstack/ev-node/block/internal/da"
	"github.com/evstack/ev-node/block/internal/executing"
	"github.com/evstack/ev-node/block/internal/pruner"
	"github.com/evstack/ev-node/block/internal/reaping"
	"github.com/evstack/ev-node/block/internal/submitting"
	"github.com/evstack/ev-node/block/internal/syncing"
	coreexecutor "github.com/evstack/ev-node/core/execution"
	coresequencer "github.com/evstack/ev-node/core/sequencer"
	"github.com/evstack/ev-node/pkg/config"
	"github.com/evstack/ev-node/pkg/genesis"
	"github.com/evstack/ev-node/pkg/signer"
	"github.com/evstack/ev-node/pkg/store"
	"github.com/evstack/ev-node/pkg/sync"
	"github.com/evstack/ev-node/pkg/telemetry"
	"github.com/evstack/ev-node/types"
)

// Components represents the block-related components
type Components struct {
	Executor  *executing.Executor
	Pruner    *pruner.Pruner
	Reaper    *reaping.Reaper
	Syncer    *syncing.Syncer
	Submitter *submitting.Submitter
	Cache     cache.Manager

	// Error channel for critical failures that should stop the node
	errorCh chan error
}

// Start starts all components and monitors for critical errors.
// It is blocking and returns when the context is cancelled or an error occurs
func (bc *Components) Start(ctx context.Context) error {
	ctxWithCancel, cancel := context.WithCancel(ctx)

	// error monitoring goroutine
	criticalErrCh := make(chan error, 1)
	go func() {
		select {
		case err := <-bc.errorCh:
			criticalErrCh <- err
			cancel()
		case <-ctx.Done():
			return
		}
	}()

	if bc.Executor != nil {
		if err := bc.Executor.Start(ctxWithCancel); err != nil {
			return fmt.Errorf("failed to start executor: %w", err)
		}
	}
	if bc.Pruner != nil {
		if err := bc.Pruner.Start(ctxWithCancel); err != nil {
			return fmt.Errorf("failed to start pruner: %w", err)
		}
	}
	if bc.Reaper != nil {
		if err := bc.Reaper.Start(ctxWithCancel); err != nil {
			return fmt.Errorf("failed to start reaper: %w", err)
		}
	}
	if bc.Syncer != nil {
		if err := bc.Syncer.Start(ctxWithCancel); err != nil {
			return fmt.Errorf("failed to start syncer: %w", err)
		}
	}
	if bc.Submitter != nil {
		if err := bc.Submitter.Start(ctxWithCancel); err != nil {
			return fmt.Errorf("failed to start submitter: %w", err)
		}
	}

	// wait for context cancellation (either from parent or critical error)
	<-ctxWithCancel.Done()

	// if we got here due to a critical error, return that error
	select {
	case err := <-criticalErrCh:
		return fmt.Errorf("node stopped due to critical execution client failure: %w", err)
	default:
		return ctx.Err()
	}
}

// Stop stops all components
func (bc *Components) Stop() error {
	var errs error
	if bc.Executor != nil {
		if err := bc.Executor.Stop(); err != nil {
			errs = errors.Join(errs, fmt.Errorf("failed to stop executor: %w", err))
		}
	}
	if bc.Pruner != nil {
		if err := bc.Pruner.Stop(); err != nil {
			errs = errors.Join(errs, fmt.Errorf("failed to stop pruner: %w", err))
		}
	}
	if bc.Reaper != nil {
		if err := bc.Reaper.Stop(); err != nil {
			errs = errors.Join(errs, fmt.Errorf("failed to stop reaper: %w", err))
		}
	}
	if bc.Syncer != nil {
		if err := bc.Syncer.Stop(); err != nil {
			errs = errors.Join(errs, fmt.Errorf("failed to stop syncer: %w", err))
		}
	}
	if bc.Submitter != nil {
		if err := bc.Submitter.Stop(); err != nil {
			errs = errors.Join(errs, fmt.Errorf("failed to stop submitter: %w", err))
		}
	}
	if bc.Cache != nil {
		if err := bc.Cache.SaveToStore(); err != nil {
			errs = errors.Join(errs, fmt.Errorf("failed to save caches: %w", err))
		}
	}

	return errs
}

// NewSyncComponents creates components for a non-aggregator full node that can only sync blocks.
// Non-aggregator full nodes can sync from P2P and DA but cannot produce blocks or submit to DA.
// They have more sync capabilities than light nodes but no block production. No signer required.
func NewSyncComponents(
	config config.Config,
	genesis genesis.Genesis,
	store store.Store,
	exec coreexecutor.Executor,
	daClient da.Client,
	headerStore header.Store[*types.P2PSignedHeader],
	dataStore header.Store[*types.P2PData],
	headerDAHintAppender submitting.DAHintAppender,
	dataDAHintAppender submitting.DAHintAppender,
	logger zerolog.Logger,
	metrics *Metrics,
	blockOpts BlockOptions,
	raftNode common.RaftNode,
) (*Components, error) {
	cacheManager, err := cache.NewManager(config, store, logger)
	if err != nil {
		return nil, fmt.Errorf("failed to create cache manager: %w", err)
	}

	// error channel for critical failures
	errorCh := make(chan error, 1)

	syncer := syncing.NewSyncer(
		store,
		exec,
		daClient,
		cacheManager,
		metrics,
		config,
		genesis,
		headerStore,
		dataStore,
		logger,
		blockOpts,
		errorCh,
		raftNode,
	)

	if config.Instrumentation.IsTracingEnabled() {
		syncer.SetBlockSyncer(syncing.WithTracingBlockSyncer(syncer))
	}

	var execPruner coreexecutor.ExecPruner
	if p, ok := exec.(coreexecutor.ExecPruner); ok {
		execPruner = p
	}
	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration, config.DA.Address)

	// Create submitter for sync nodes (no signer, only DA inclusion processing)
	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerDAHintAppender, dataDAHintAppender)
	if config.Instrumentation.IsTracingEnabled() {
		daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
	}
	submitter := submitting.NewSubmitter(
		store,
		exec,
		cacheManager,
		metrics,
		config,
		genesis,
		daSubmitter,
		nil, // No sequencer for sync nodes
		nil, // No signer for sync nodes
		logger,
		errorCh,
	)

	return &Components{
		Syncer:    syncer,
		Submitter: submitter,
		Cache:     cacheManager,
		Pruner:    pruner,
		errorCh:   errorCh,
	}, nil
}

// NewAggregatorComponents creates components for an aggregator full node that can produce and sync blocks.
// Aggregator nodes have full capabilities - they can produce blocks, sync from P2P and DA,
// and submit headers/data to DA. Requires a signer for block production and DA submission.
func NewAggregatorComponents(
	config config.Config,
	genesis genesis.Genesis,
	store store.Store,
	exec coreexecutor.Executor,
	sequencer coresequencer.Sequencer,
	daClient da.Client,
	signer signer.Signer,
	headerSyncService *sync.HeaderSyncService,
	dataSyncService *sync.DataSyncService,
	logger zerolog.Logger,
	metrics *Metrics,
	blockOpts BlockOptions,
	raftNode common.RaftNode,
) (*Components, error) {
	cacheManager, err := cache.NewManager(config, store, logger)
	if err != nil {
		return nil, fmt.Errorf("failed to create cache manager: %w", err)
	}

	// error channel for critical failures
	errorCh := make(chan error, 1)

	// wrap sequencer with tracing if enabled
	if config.Instrumentation.IsTracingEnabled() {
		sequencer = telemetry.WithTracingSequencer(sequencer)
	}

	executor, err := executing.NewExecutor(
		store,
		exec,
		sequencer,
		signer,
		cacheManager,
		metrics,
		config,
		genesis,
		headerSyncService,
		dataSyncService,
		logger,
		blockOpts,
		errorCh,
		raftNode,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create executor: %w", err)
	}

	if config.Instrumentation.IsTracingEnabled() {
		executor.SetBlockProducer(executing.WithTracingBlockProducer(executor))
	}

	var execPruner coreexecutor.ExecPruner
	if p, ok := exec.(coreexecutor.ExecPruner); ok {
		execPruner = p
	}
	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration, config.DA.Address)

	reaper, err := reaping.NewReaper(
		exec,
		sequencer,
		genesis,
		logger,
		executor,
		cacheManager,
		config.Node.ScrapeInterval.Duration,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create reaper: %w", err)
	}

	if config.Node.BasedSequencer { // no submissions needed for bases sequencer
		return &Components{
			Executor: executor,
			Pruner:   pruner,
			Reaper:   reaper,
			Cache:    cacheManager,
			errorCh:  errorCh,
		}, nil
	}

	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerSyncService, dataSyncService)
	if config.Instrumentation.IsTracingEnabled() {
		daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
	}
	submitter := submitting.NewSubmitter(
		store,
		exec,
		cacheManager,
		metrics,
		config,
		genesis,
		daSubmitter,
		sequencer,
		signer, // Signer for aggregator nodes to submit to DA
		logger,
		errorCh,
	)

	return &Components{
		Executor:  executor,
		Pruner:    pruner,
		Reaper:    reaper,
		Submitter: submitter,
		Cache:     cacheManager,
		errorCh:   errorCh,
	}, nil
}