engine.go

package dasmon

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	"github.com/ethp2p/dasmon/metrics"
	"github.com/ethp2p/dasmon/store"
	"github.com/libp2p/go-libp2p/core/event"
	"github.com/libp2p/go-libp2p/core/host"
	"github.com/libp2p/go-libp2p/core/network"
	"github.com/libp2p/go-libp2p/core/peer"
	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
	"gopkg.in/natefinch/lumberjack.v2"
)

// Package-level logger for dasmon
var log = slog.Default().WithGroup("dasmon")

type Engine struct {
	// Environment contains user-provided objects
	env *Environment

	// Config contains monitoring behavior configuration
	config *Config

	// RPC client
	rpcClient *RpcClient

	// Event system
	eventbus event.Bus
	emitter  event.Emitter

	// Event file writer (optional NDJSON log)
	eventWriter io.WriteCloser

	// Database for test history persistence
	store store.Store

	// Internal state
	ctx       context.Context
	cancel    context.CancelFunc
	running   atomic.Bool
	startTime time.Time

	// Job constraints
	jobConstraints *JobConstraints

	// Worker pool
	workers       int
	workerWg      sync.WaitGroup
	jobCh         chan *Job
	doneCh        chan *Job
	workerReadyCh chan struct{}

	// Peer tracking
	peersMu     sync.RWMutex
	peers       map[peer.ID]*MonitoredPeer
	peerList    []*MonitoredPeer
	nextPeerIdx int

	// Status refresh scheduling
	refreshQueue    []*MonitoredPeer
	refreshQueueMu  sync.Mutex
	refreshSignalCh chan struct{}

	// Metrics
	jobsTotal     atomic.Uint64
	jobsCompleted atomic.Uint64
	jobsSucceeded atomic.Uint64
	jobsFailed    atomic.Uint64
	jobsInflight  atomic.Int64

	// Chain tracking
	commTracker *CommitmentTracker
	chainHead   atomic.Uint64

	// Event sequencing for WebSocket consistency
	eventSeqNum atomic.Uint64

	// Subscriptions
	netsub      event.Subscription
	blockEvents event.Subscription

	getPeersCh chan chan []PeerCustody
}

func NewEngine(env *Environment, config *Config) (*Engine, error) {
	if err := env.Validate(); err != nil {
		return nil, fmt.Errorf("invalid dasmon environment: %w", err)
	}
	if err := config.Validate(); err != nil {
		return nil, fmt.Errorf("invalid dasmon config: %w", err)
	}

	// Initialize metrics
	metrics.Initialize(config.Metrics)

	ctx, cancel := context.WithCancel(context.Background())

	// Open and migrate database
	database := store.NewStore()
	if err := database.Open(config.DatabasePath); err != nil {
		cancel()
		return nil, fmt.Errorf("failed to open database: %w", err)
	}
	if err := database.Migrate(); err != nil {
		database.Close()
		cancel()
		return nil, fmt.Errorf("failed to migrate database: %w", err)
	}

	bus := eventbus.NewBus()
	emitter, err := bus.Emitter(new(Event))
	if err != nil {
		database.Close()
		cancel()
		return nil, fmt.Errorf("could not create emitter: %w", err)
	}

	engine := &Engine{
		env:        env,
		ctx:        ctx,
		cancel:     cancel,
		config:     config,
		peers:      make(map[peer.ID]*MonitoredPeer),
		getPeersCh: make(chan chan []PeerCustody, 1),
		rpcClient:  NewClient(env.Host),
		eventbus:   bus,
		emitter:    emitter,
		store:      database,
	}

	// Initialize event file writer if configured
	if config.EventLogFile != "" {
		engine.eventWriter = &lumberjack.Logger{
			Filename:  config.EventLogFile,
			MaxSize:   100, // megabytes
			MaxAge:    30,  // days
			LocalTime: true,
			Compress:  true,
		}
		log.Info("event file writer enabled", "path", config.EventLogFile)
	}

	return engine, nil
}

// Start begins monitoring with the given policy
func (e *Engine) Start() (err error) {
	if !e.running.CompareAndSwap(false, true) {
		return fmt.Errorf("engine already running")
	}

	e.startTime = time.Now()

	// Subscribe to libp2p network events
	e.netsub, err = e.Host().EventBus().Subscribe(new(event.EvtPeerConnectednessChanged))
	if err != nil {
		_ = e.emitter.Close()
		e.cancel()
		return fmt.Errorf("could not subscribe to event bus: %w", err)
	}

	head, err := e.loadHeadBlock()
	if err != nil {
		return fmt.Errorf("failed to load head block: %w", err)
	}

	e.chainHead.Store(head.Slot)

	// Load our initial peers
	e.loadInitialPeers()

	// Initialize peer count metric
	metrics.PeersConnected.Set(float64(len(e.peers)))

	e.commTracker = NewCommitmentTracker(e.config.BlockFilter, head)

	var fetcher Fetcher = NewFetcherDB(e.env.Externs)
	if e.config.Backfill.BeaconAPI != "" {
		beacon := NewFetcherBeaconAPI(e.config.Backfill.BeaconAPI)
		fetcher = ChainedFetcher{fetcher, beacon}
	}

	switch e.config.Backfill.Mode {
	case BackfillModePreload:
		if err := e.commTracker.Backfill(e.ctx, fetcher); err != nil {
			return fmt.Errorf("failed to backfill commitments: %w", err)
		}
	case BackfillModeBackground:
		go e.commTracker.Backfill(e.ctx, fetcher)
	default:
		return fmt.Errorf("invalid backfill mode: %s", e.config.Backfill.Mode)
	}

	sampler, err := NewSampler(e.config.Sampling)
	if err != nil {
		return fmt.Errorf("invalid sampling config: %w", err)
	}

	e.jobConstraints = &JobConstraints{
		MaxColumns:        e.config.Sampling.MaxColumns,
		MaxSlots:          e.config.Sampling.MaxSlots,
		BlockFilter:       e.config.BlockFilter,
		Sampler:           sampler,
		CommitmentTracker: e.commTracker,
		TestInterval:      e.config.TestInterval,
	}

	// Initialize channels
	e.workers = e.config.Workers
	e.jobCh = make(chan *Job, e.workers)
	e.doneCh = make(chan *Job, 32)
	e.workerReadyCh = make(chan struct{}, e.workers)
	e.refreshSignalCh = make(chan struct{}, 1)

	// Start worker pool
	for i := 0; i < e.workers; i++ {
		e.workerWg.Add(1)
		go e.worker(i)
	}

	// Start dispatch loop
	e.workerWg.Add(1)
	go e.dispatchLoop()

	// Start status check scheduler
	e.workerWg.Add(1)
	go e.statusCheckScheduler()

	// Start tracking loop
	e.workerWg.Add(1)
	go e.trackingLoop()

	// Start commitment tracker
	e.workerWg.Add(1)
	go e.chainTracker()

	// Start event processor
	e.workerWg.Add(1)
	go e.eventProcessor()

	// Log startup configuration
	log.Info("dasmon engine started",
		"workers", e.workers,
		"test_interval", time.Duration(e.config.TestInterval),
		"status_refresh", time.Duration(e.config.PeerRefreshInterval),
		"reconnection_grace", time.Duration(e.config.ReconnectionGrace),
		"max_columns", e.config.Sampling.MaxColumns,
		"slot_bias", e.config.Sampling.SlotBias,
		"prefer_untested", e.config.Sampling.PreferUntested,
		"history_window", time.Duration(e.config.Sampling.HistoryWindow),
		"request_slots_max", e.config.Sampling.MaxSlots,
		"block_filter_start", e.config.BlockFilter.StartSlot,
		"block_filter_end", e.config.BlockFilter.EndSlot,
		"include_peers", len(e.config.NodeFilter.Include),
		"exclude_peers", len(e.config.NodeFilter.Exclude),
		"metrics_enabled", e.config.Metrics.Enabled,
		"commitment_buffer", MaxCustodySlots,
	)

	return nil
}

func (e *Engine) loadHeadBlock() (BlockCommitmentInfo, error) {
	block, root, err := e.env.Externs.ChainHead(e.ctx)
	if err != nil {
		return BlockCommitmentInfo{}, fmt.Errorf("failed to get chain head: %w", err)
	}

	return BlockCommitmentInfo{
		Slot:            block.Slot,
		BlockRoot:       root,
		BlobCommitments: block.BlobCommitments,
	}, nil
}

func (e *Engine) loadInitialPeers() {
	peers := e.Host().Network().Peers()

	for _, pid := range peers {
		mp, err := newMonitoredPeer(pid, e.store)
		if err != nil {
			log.Warn("failed to create monitored peer", "peer", pid, "error", err)
			continue
		}
		mp.Context, mp.CancelFunc = context.WithCancel(e.ctx)

		// Initialize peer (loads history, fetches status) - blocks until complete
		// Peer is only added to engine state after initialization succeeds
		e.workerWg.Add(1)
		go e.initializePeerWithHistory(mp)
	}
}

func (e *Engine) Subscribe(bufferSize int) (event.Subscription, error) {
	return e.eventbus.Subscribe(new(Event), eventbus.BufSize(bufferSize))
}

// Stop gracefully stops the Engine
func (e *Engine) Stop() error {
	if !e.running.CompareAndSwap(true, false) {
		return fmt.Errorf("engine not running")
	}

	// Cancel context to stop all goroutines
	if e.cancel != nil {
		e.cancel()
	}

	// Close subscriptions
	if e.netsub != nil {
		_ = e.netsub.Close()
	}

	if e.blockEvents != nil {
		_ = e.blockEvents.Close()
	}

	// Close channels to signal workers
	if e.jobCh != nil {
		close(e.jobCh)
	}

	// Wait for all goroutines to finish
	e.workerWg.Wait()

	// Close emitter
	if e.emitter != nil {
		_ = e.emitter.Close()
	}

	// Close event writer
	if e.eventWriter != nil {
		if err := e.eventWriter.Close(); err != nil {
			log.Warn("failed to close event writer", "error", err)
		}
	}

	// Close database
	if e.store != nil {
		if err := e.store.Close(); err != nil {
			log.Warn("failed to close database", "error", err)
			return fmt.Errorf("failed to close database: %w", err)
		}
	}

	// Clear state
	e.cancel = nil
	e.ctx = nil
	e.netsub = nil
	e.blockEvents = nil
	e.emitter = nil
	e.eventbus = nil
	e.store = nil

	return nil
}

// TODO when connecting to peers, put them through the allowlist

func (e *Engine) trackingLoop() {
	defer e.workerWg.Done()

	cleanupTicker := time.NewTicker(1 * time.Minute)
	defer cleanupTicker.Stop()

	for {
		select {
		case ev, ok := <-e.netsub.Out():
			if !ok {
				return
			}

			evt := ev.(event.EvtPeerConnectednessChanged)
			logger := log.With("peer", evt.Peer, "status", evt.Connectedness)

			switch evt.Connectedness {
			case network.Connected:
				e.peersMu.Lock()
				mp, exists := e.peers[evt.Peer]

				if exists {
					// Reconnection within grace period
					logger.Info("peer reconnected within grace period")
					mp.MarkReconnected()

					// Reschedule immediate refresh
					e.peersMu.Unlock()
					e.scheduleRefresh(mp, 0)
				} else {
					// New peer
					logger.Debug("tracking new peer")
					e.peersMu.Unlock()

					mp, err := newMonitoredPeer(evt.Peer, e.store)
					if err != nil {
						logger.Warn("failed to create monitored peer", "error", err)
						continue
					}
					mp.Context, mp.CancelFunc = context.WithCancel(e.ctx)

					metrics.PeerEventsTotal.WithLabelValues("connected").Inc()

					// Initialize asynchronously (will add to engine state after loading history)
					e.workerWg.Add(1)
					go e.initializePeerWithHistory(mp)
				}

			case network.NotConnected:
				e.peersMu.RLock()
				mp, ok := e.peers[evt.Peer]
				e.peersMu.RUnlock()

				if ok {
					logger.Debug("peer disconnected")
					now := time.Now()
					mp.MarkDisconnected(now)
					mp.CancelFunc() // Cancel any in-flight operations

					metrics.PeerEventsTotal.WithLabelValues("disconnected").Inc()
				}
			}

		case <-cleanupTicker.C:
			e.cleanupDisconnectedPeers()

		case respCh := <-e.getPeersCh:
			e.peersMu.RLock()
			result := make([]PeerCustody, 0, len(e.peers))
			for id, mp := range e.peers {
				custody := mp.Custody()
				custody.PeerID = id
				result = append(result, custody)
			}
			e.peersMu.RUnlock()
			respCh <- result

		case <-e.ctx.Done():
			return
		}
	}
}

// eventProcessor consumes completed jobs and emits events
func (e *Engine) eventProcessor() {
	defer e.workerWg.Done()

	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case job := <-e.doneCh:
			e.processCompletedJob(job)

			// Update metrics
			e.jobsCompleted.Add(1)
			if job.Result.Error == nil {
				e.jobsSucceeded.Add(1)
			} else {
				e.jobsFailed.Add(1)
			}

		case <-ticker.C:
			// Emit periodic snapshot event
			status := e.getStatus()
			evt := &EvtSystemSnapshot{
				CurrentHeadSlot: e.chainHead.Load(),
				Status:          status,
			}
			e.emitEvent(Event{Snapshot: evt})

		case <-e.ctx.Done():
			return
		}
	}
}

// emitEvent emits an event to subscribers and optionally writes to file
func (m *Engine) emitEvent(evt Event) {
	// Assign sequence number and timestamp atomically
	evt.SeqNum = m.eventSeqNum.Add(1)
	evt.Timestamp = time.Now()

	// Write to event file if configured
	if m.eventWriter != nil {
		data, err := json.Marshal(evt)
		if err != nil {
			log.Warn("failed to marshal event for file writer", "error", err)
		} else {
			if _, err := m.eventWriter.Write(append(data, '\n')); err != nil {
				log.Warn("failed to write event to file", "error", err)
			}
		}
	}

	// Emit to subscribers
	if err := m.emitter.Emit(evt); err != nil {
		log.Warn("failed to emit event", "error", err)
	}
}

// chainTracker monitors for new blocks and updates the commitment ring buffer
func (e *Engine) chainTracker() {
	defer e.workerWg.Done()

	log.Info("starting chain tracker")

	// Subscribe to chain events
	eventChan, err := e.env.Externs.ChainSubscribe(e.ctx)
	if err != nil {
		log.Error("failed to subscribe to chain events", "error", err)
		return
	}

	last := e.chainHead.Load()

	log.Info("commitment tracker subscribed to head change events")

	for {
		select {
		case ev, ok := <-eventChan:
			if !ok {
				log.Info("chain event channel closed")
				return
			}

			if ev.Type != ChainEventNewHead {
				continue
			}

			current := ev.Slot
			if current < last {
				log.Warn("detected chain reorganization", "from_slot", last, "to_slot", current)
				continue
			}

			// Query blocks in range [last+1, current]
			blocks, err := e.env.Externs.ChainQuery(e.ctx, BlockQuery{
				StartSlot: last + 1,
				EndSlot:   current,
			})
			if err != nil {
				log.Warn("failed to get blocks for slots", "start", last+1, "end", current, "error", err)
				continue
			}

			for _, block := range blocks {
				e.commTracker.Append(block.Slot, block.Root, block.BlobCommitments)
			}

			e.chainHead.Store(current)
			last = current

			log.Debug("updated commitment tracker head via event", "head_slot", current)

		case <-e.ctx.Done():
			log.Info("stopping commitment tracker")
			return
		}
	}
}

func (e *Engine) Host() host.Host {
	return e.env.Host
}

// Status returns the current engine status for the HTTP interface
func (e *Engine) Status(ctx context.Context) (*EngineStatus, error) {
	return e.getStatus(), nil
}

// Running returns whether the engine is currently running
func (e *Engine) Running() bool {
	return e.running.Load()
}

// StartTime returns when the engine was started
func (e *Engine) StartTime() time.Time {
	return e.startTime
}

// EngineStatus represents the current status of the engine
type EngineStatus struct {
	PeersTracked  int
	JobsPending   int
	JobsInflight  int
	JobsFinished  int
	JobsSucceeded int
	JobsFailed    int
	JobsTotal     int
	Workers       int
}

// GetMonitoredPeers returns custody information for all currently monitored peers
// This method sends a request to the tracking loop to avoid locks
func (e *Engine) GetMonitoredPeers(ctx context.Context) ([]PeerCustody, error) {
	ch := make(chan []PeerCustody, 1)
	select {
	case e.getPeersCh <- ch:
	case <-ctx.Done():
		return nil, ctx.Err()
	}

	select {
	case result := <-ch:
		return result, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

// GetSnapshot returns a consistent snapshot of the engine state along with
// the sequence number at which the snapshot was captured. This ensures that
// WebSocket clients can subscribe to events starting from this sequence number
// without missing any events.
func (e *Engine) GetSnapshot(ctx context.Context) (*Snapshot, error) {
	// Lock to ensure consistency between state and sequence number
	e.peersMu.RLock()

	// Capture sequence number while holding lock
	seqNum := e.eventSeqNum.Load()

	// Gather peer custody info
	peers := make([]PeerCustody, 0, len(e.peers))
	for id, mp := range e.peers {
		custody := mp.Custody()
		custody.PeerID = id
		peers = append(peers, custody)
	}
	e.peersMu.RUnlock()

	// Get status (uses its own locking)
	status := e.getStatus()

	uptime := int64(0)
	if !e.startTime.IsZero() {
		uptime = int64(time.Since(e.startTime).Seconds())
	}

	snapshot := &Snapshot{
		SeqNum:     seqNum,
		Status:     status,
		Peers:      peers,
		HeadSlot:   e.chainHead.Load(),
		Config:     e.config,
		Running:    e.running.Load(),
		StartTime:  e.startTime.Unix(),
		UptimeSecs: uptime,
	}

	return snapshot, nil
}

// GetCurrentHead returns the current chain head slot
func (e *Engine) GetCurrentHead() uint64 {
	return e.chainHead.Load()
}

// worker pulls jobs when ready and executes them
func (e *Engine) worker(id int) {
	defer e.workerWg.Done()

	wlog := log.With("worker", id)
	wlog.Debug("worker started")
	defer wlog.Debug("worker stopped")

	for {
		// Signal ready for work
		select {
		case e.workerReadyCh <- struct{}{}:
		case <-e.ctx.Done():
			return
		}

		// Wait for job assignment
		select {
		case job := <-e.jobCh:
			if job == nil {
				return // Shutdown signal
			}

			wlog.Debug("received job", "job_id", job.ID)

			// Create RPC client and execute
			client := NewClient(e.Host())
			e.jobsInflight.Add(1)

			err := job.Execute(client)
			if err != nil {
				job.log.Error("job execution failed", "error", err)
				job.Result.Error = err
			} else {
				job.log.Debug("job execution succeeded")
			}

			e.jobsInflight.Add(-1)

			// Signal completion
			select {
			case e.doneCh <- job:
			case <-e.ctx.Done():
				return
			}

		case <-e.ctx.Done():
			return
		}
	}
}

// dispatchLoop waits for workers to be ready, then selects and dispatches jobs
func (e *Engine) dispatchLoop() {
	defer e.workerWg.Done()

	log.Debug("dispatch loop started")
	defer log.Debug("dispatch loop stopped")

	for {
		select {
		case <-e.workerReadyCh:
			job := e.selectJob()
			if job == nil {
				// No job ready after full round-robin, put worker token back and backoff
				e.workerReadyCh <- struct{}{}

				// Brief sleep to avoid busy loop when no work available
				select {
				case <-time.After(10 * time.Millisecond):
				case <-e.ctx.Done():
					return
				}
				continue
			}

			// Prepare job for execution
			job.Prepare(e.ctx)
			e.jobsTotal.Add(1)

			// Send to waiting worker
			select {
			case e.jobCh <- job:
			case <-e.ctx.Done():
				job.Cancel()
				return
			}

		case <-e.ctx.Done():
			return
		}
	}
}

// selectJob does round-robin over peers, returns first ready job
func (e *Engine) selectJob() *Job {
	e.peersMu.RLock()
	defer e.peersMu.RUnlock()

	if len(e.peerList) == 0 {
		return nil
	}

	// Round-robin through all peers once
	for i := 0; i < len(e.peerList); i++ {
		// Reshuffle our peerlist if we've cycled through all peers.
		if e.nextPeerIdx == 0 {
			rand.Shuffle(len(e.peers), func(i, j int) {
				e.peerList[i], e.peerList[j] = e.peerList[j], e.peerList[i]
			})
		}

		mp := e.peerList[e.nextPeerIdx]
		e.nextPeerIdx = (e.nextPeerIdx + 1) % len(e.peerList)

		if mp.IsDisconnected() {
			continue
		}

		if job := mp.PollJob(e.jobConstraints); job != nil {
			return job
		}
	}

	// Full round-robin completed with no jobs
	return nil
}

// statusCheckScheduler manages the refresh queue and triggers status checks
func (e *Engine) statusCheckScheduler() {
	defer e.workerWg.Done()

	log.Debug("status check scheduler started")
	defer log.Debug("status check scheduler stopped")

	for {
		e.refreshQueueMu.Lock()

		if len(e.refreshQueue) == 0 {
			// No peers to refresh, wait for signal
			e.refreshQueueMu.Unlock()

			select {
			case <-e.refreshSignalCh:
				continue // Check queue again
			case <-e.ctx.Done():
				return
			}
		}

		// Get first peer and check when it needs refresh
		peer := e.refreshQueue[0]
		nextRefresh := peer.GetNextRefreshAt()
		e.refreshQueueMu.Unlock()

		// Wait until it's time to refresh
		waitDuration := time.Until(nextRefresh)
		if waitDuration < 0 {
			waitDuration = 0
		}

		timer := time.NewTimer(waitDuration)

		select {
		case <-timer.C:
			// Time to refresh this peer
			e.refreshQueueMu.Lock()
			if len(e.refreshQueue) > 0 && e.refreshQueue[0] == peer {
				// Remove from queue
				e.refreshQueue = e.refreshQueue[1:]
			}
			e.refreshQueueMu.Unlock()

			// Launch refresh in goroutine
			e.workerWg.Add(1)
			go e.refreshPeer(peer)

		case <-e.refreshSignalCh:
			// New peer added or queue changed, recheck
			timer.Stop()

		case <-e.ctx.Done():
			timer.Stop()
			return
		}
	}
}

// refreshPeer fetches status/metadata and updates custody, then reschedules
func (e *Engine) refreshPeer(mp *MonitoredPeer) {
	defer e.workerWg.Done()

	// Check if peer is disconnected
	if mp.IsDisconnected() {
		mp.log.Debug("skipping refresh for disconnected peer")
		return
	}

	peerInfo, err := e.env.Externs.PeerRefresh(e.ctx, mp.peerId)
	if err != nil {
		mp.log.Warn("error refreshing peer info", "error", err)
		e.scheduleRefresh(mp, time.Duration(e.config.PeerRefreshInterval))
		return
	}

	if peerInfo == nil {
		mp.log.Debug("peer has no metadata or status")
		e.scheduleRefresh(mp, time.Duration(e.config.PeerRefreshInterval))
		return
	}

	// Update custody
	prev, next := mp.refresh(peerInfo)

	// Emit custody update event if changed
	if !prev.Equals(next) {
		head := e.chainHead.Load()
		evt := &EvtCustodyUpdate{
			PeerID:        &mp.peerId,
			DeltaFromHead: int64(head) - int64(next.EarliestSlot),
		}

		if prev.CustodyGroupCount != next.CustodyGroupCount {
			vc := &ValueChange[uint64]{prev.CustodyGroupCount, next.CustodyGroupCount}
			evt.Cgc = vc
		}

		if prev.EarliestSlot != next.EarliestSlot {
			vc := &ValueChange[uint64]{uint64(prev.EarliestSlot), uint64(next.EarliestSlot)}
			evt.EarliestSlot = vc
		}

		e.emitEvent(Event{Custody: evt})
	}

	// Reschedule next refresh
	e.scheduleRefresh(mp, time.Duration(e.config.PeerRefreshInterval))
}

// initializePeer fetches initial status/metadata for a new peer
func (e *Engine) initializePeerWithHistory(mp *MonitoredPeer) {
	defer e.workerWg.Done()

	// Load test history from database (within history window)
	historyWindow := time.Now().Add(-time.Duration(e.config.Sampling.HistoryWindow))
	if err := mp.history.LoadSince(e.ctx, historyWindow); err != nil {
		mp.log.Error("failed to load peer history, aborting peer initialization", "error", err)
		mp.CancelFunc()
		return
	}

	stats := mp.history.Stats()
	mp.log.Debug("loaded peer history",
		"history_slots", stats.SlotsTracked,
		"history_tests", stats.TotalTests,
		"history_memory", stats.MemoryBytes,
	)

	// Add random initial delay to stagger status checks
	stagger := time.Duration(rand.Int63n(int64(e.config.PeerRefreshInterval)))

	select {
	case <-time.After(stagger):
	case <-e.ctx.Done():
		return
	case <-mp.Done():
		return
	}

	peerInfo, err := e.env.Externs.PeerRefresh(e.ctx, mp.peerId)
	if err != nil {
		mp.log.Warn("error getting initial peer info", "error", err)
		e.scheduleRefresh(mp, time.Duration(e.config.PeerRefreshInterval))
		return
	}

	if peerInfo == nil {
		mp.log.Info("peer has no metadata or status; will retry")
		e.scheduleRefresh(mp, time.Duration(e.config.PeerRefreshInterval))
		return
	}

	// Initialize custody
	mp.refresh(peerInfo)

	// Set initial nextRunAt with random stagger
	mp.mu.Lock()
	initialJobDelay := time.Duration(rand.Int63n(int64(e.config.TestInterval)))
	mp.nextRunAt = time.Now().Add(initialJobDelay)
	mp.mu.Unlock()

	// Add to engine state (AFTER successful initialization)
	e.peersMu.Lock()
	e.peers[mp.peerId] = mp
	e.peerList = append(e.peerList, mp)
	e.peersMu.Unlock()

	mp.log.Info("peer initialized",
		"custody", mp.custody.String(),
		"next_job_in", initialJobDelay,
	)

	// Schedule first refresh
	e.scheduleRefresh(mp, time.Duration(e.config.PeerRefreshInterval))

	// Emit tracking event
	e.emitEvent(Event{Tracking: &EvtNodeTracking{PeerID: &mp.peerId, Tracking: true}})
}

// scheduleRefresh adds peer to refresh queue and signals scheduler
func (e *Engine) scheduleRefresh(mp *MonitoredPeer, interval time.Duration) {
	nextRefresh := time.Now().Add(interval)
	mp.SetNextRefreshAt(nextRefresh)

	e.refreshQueueMu.Lock()
	e.refreshQueue = append(e.refreshQueue, mp)
	e.refreshQueueMu.Unlock()

	// Signal scheduler (non-blocking)
	select {
	case e.refreshSignalCh <- struct{}{}:
	default:
	}
}

// getStatus returns current engine status
func (e *Engine) getStatus() *EngineStatus {
	e.peersMu.RLock()
	peerCount := len(e.peers)
	e.peersMu.RUnlock()

	total := e.jobsTotal.Load()
	completed := e.jobsCompleted.Load()
	succeeded := e.jobsSucceeded.Load()
	failed := e.jobsFailed.Load()
	inflight := e.jobsInflight.Load()
	pending := total - completed - uint64(inflight)

	return &EngineStatus{
		PeersTracked:  peerCount,
		JobsPending:   int(pending),
		JobsInflight:  int(inflight),
		JobsFinished:  int(completed),
		JobsSucceeded: int(succeeded),
		JobsFailed:    int(failed),
		JobsTotal:     int(total),
		Workers:       e.workers,
	}
}

// cleanupDisconnectedPeers removes peers that have been disconnected beyond grace period
func (e *Engine) cleanupDisconnectedPeers() {
	e.peersMu.Lock()
	defer e.peersMu.Unlock()

	grace := time.Duration(e.config.ReconnectionGrace)
	now := time.Now()

	for pid, mp := range e.peers {
		mp.mu.RLock()
		disconnectedAt := mp.disconnectedAt
		mp.mu.RUnlock()

		if disconnectedAt != nil && now.Sub(*disconnectedAt) > grace {
			// Remove from map
			delete(e.peers, pid)

			// Remove from round-robin list
			for i, p := range e.peerList {
				if p == mp {
					e.peerList = append(e.peerList[:i], e.peerList[i+1:]...)
					if e.nextPeerIdx > i {
						e.nextPeerIdx--
					}
					break
				}
			}

			// Remove from refresh queue (scan and remove all occurrences)
			e.refreshQueueMu.Lock()
			newQueue := make([]*MonitoredPeer, 0, len(e.refreshQueue))
			for _, p := range e.refreshQueue {
				if p != mp {
					newQueue = append(newQueue, p)
				}
			}
			e.refreshQueue = newQueue
			e.refreshQueueMu.Unlock()

			log.Debug("removed disconnected peer after grace period", "peer", pid)

			evt := &EvtNodeTracking{
				PeerID:     &pid,
				Dropped:    true,
				DropReason: "peer disconnected (grace period expired)",
			}
			e.emitEvent(Event{Tracking: evt})

			metrics.PeersConnected.Set(float64(len(e.peers)))
		}
	}
}

// SlotResult contains the result for a specific slot
type SlotResult struct {
	Slot            uint64
	BlockHash       string
	ColumnsExpected []uint64
	ColumnsFound    []uint64
	ColumnsMissing  []uint64
	Success         bool
	Error           error
	Duration        time.Duration
}

// TestSummary provides aggregate statistics
type TestSummary struct {
	TotalSlots       int
	SuccessfulSlots  int
	FailedSlots      int
	TotalColumns     int
	FoundColumns     int
	MissingColumns   int
	AverageLatency   time.Duration
	CustodyFulfilled bool
}

// Stats represents the current state of the Engine
type Stats struct {
	Running         bool
	StartTime       time.Time
	NodesTracked    int
	TestsCompleted  int
	TestsInProgress int
	LastTestTime    time.Time
	CurrentHead     uint64
	TotalEvents     uint64
}