mev-beta/orig/test/mock_sequencer_service.go

//go:build integration && legacy && forked
// +build integration,legacy,forked

package test_main

import (
	"context"
	"encoding/json"
	"fmt"
	"math/rand"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"

	"github.com/ethereum/go-ethereum/common"

	"github.com/fraktal/mev-beta/internal/logger"
	"github.com/fraktal/mev-beta/pkg/arbitrum"
)

// MockSequencerService simulates Arbitrum sequencer behavior for testing
type MockSequencerService struct {
	config  *SequencerConfig
	logger  *logger.Logger
	storage *TransactionStorage

	// Simulation state
	currentBlock     uint64
	transactionQueue []*SequencerTransaction
	subscribers      map[string]chan *SequencerBlock

	// Timing simulation
	blockTimer *time.Ticker
	batchTimer *time.Ticker

	// Control
	running bool
	mu      sync.RWMutex

	// Metrics
	metrics *SequencerMetrics
}

// SequencerTransaction represents a transaction in the sequencer
type SequencerTransaction struct {
	*RealTransactionData

	// Sequencer-specific fields
	SubmittedAt     time.Time     `json:"submitted_at"`
	ProcessedAt     time.Time     `json:"processed_at"`
	BatchID         string        `json:"batch_id"`
	SequenceNumber  uint64        `json:"sequence_number"`
	Priority        int           `json:"priority"`
	CompressionSize int           `json:"compression_size"`
	ValidationTime  time.Duration `json:"validation_time"`
	InclusionDelay  time.Duration `json:"inclusion_delay"`
}

// SequencerBlock represents a block produced by the sequencer
type SequencerBlock struct {
	Number           uint64                  `json:"number"`
	Hash             common.Hash             `json:"hash"`
	ParentHash       common.Hash             `json:"parent_hash"`
	Timestamp        time.Time               `json:"timestamp"`
	Transactions     []*SequencerTransaction `json:"transactions"`
	BatchID          string                  `json:"batch_id"`
	GasUsed          uint64                  `json:"gas_used"`
	GasLimit         uint64                  `json:"gas_limit"`
	CompressionRatio float64                 `json:"compression_ratio"`
	ProcessingTime   time.Duration           `json:"processing_time"`
}

// SequencerMetrics tracks sequencer performance metrics
type SequencerMetrics struct {
	BlocksProduced          uint64        `json:"blocks_produced"`
	TransactionsProcessed   uint64        `json:"transactions_processed"`
	DEXTransactionsFound    uint64        `json:"dex_transactions_found"`
	MEVTransactionsFound    uint64        `json:"mev_transactions_found"`
	AverageBlockTime        time.Duration `json:"average_block_time"`
	AverageTxPerBlock       float64       `json:"average_tx_per_block"`
	AverageCompressionRatio float64       `json:"average_compression_ratio"`
	TotalProcessingTime     time.Duration `json:"total_processing_time"`
	ErrorCount              uint64        `json:"error_count"`

	// Real-time metrics
	LastBlockTime   time.Time `json:"last_block_time"`
	QueueSize       int       `json:"queue_size"`
	SubscriberCount int       `json:"subscriber_count"`

	mu sync.RWMutex
}

// NewMockSequencerService creates a new mock sequencer service
func NewMockSequencerService(config *SequencerConfig, logger *logger.Logger, storage *TransactionStorage) *MockSequencerService {
	return &MockSequencerService{
		config:      config,
		logger:      logger,
		storage:     storage,
		subscribers: make(map[string]chan *SequencerBlock),
		metrics:     &SequencerMetrics{},
	}
}

// Start starts the mock sequencer service
func (mss *MockSequencerService) Start(ctx context.Context) error {
	mss.mu.Lock()
	defer mss.mu.Unlock()

	if mss.running {
		return fmt.Errorf("sequencer service already running")
	}

	mss.logger.Info("Starting mock Arbitrum sequencer service...")

	// Initialize state
	mss.currentBlock = mss.config.StartBlock
	if mss.currentBlock == 0 {
		mss.currentBlock = 150000000 // Start from recent Arbitrum block
	}

	// Load test data from storage
	if err := mss.loadTestData(); err != nil {
		return fmt.Errorf("failed to load test data: %w", err)
	}

	// Start block production timer
	mss.blockTimer = time.NewTicker(mss.config.SequencerTiming)

	// Start batch processing timer (faster than blocks)
	batchInterval := mss.config.SequencerTiming / 4 // 4 batches per block
	mss.batchTimer = time.NewTicker(batchInterval)

	mss.running = true

	// Start goroutines
	go mss.blockProductionLoop(ctx)
	go mss.batchProcessingLoop(ctx)
	go mss.metricsUpdateLoop(ctx)

	mss.logger.Info(fmt.Sprintf("Mock sequencer started - producing blocks every %v", mss.config.SequencerTiming))
	return nil
}

// Stop stops the mock sequencer service
func (mss *MockSequencerService) Stop() {
	mss.mu.Lock()
	defer mss.mu.Unlock()

	if !mss.running {
		return
	}

	mss.logger.Info("Stopping mock sequencer service...")

	mss.running = false

	if mss.blockTimer != nil {
		mss.blockTimer.Stop()
	}
	if mss.batchTimer != nil {
		mss.batchTimer.Stop()
	}

	// Close subscriber channels
	for id, ch := range mss.subscribers {
		close(ch)
		delete(mss.subscribers, id)
	}

	mss.logger.Info("Mock sequencer stopped")
}

// loadTestData loads transaction data from storage for simulation
func (mss *MockSequencerService) loadTestData() error {
	// Get recent transactions from storage
	stats := mss.storage.GetStorageStats()
	if stats.TotalTransactions == 0 {
		mss.logger.Warn("No test data available in storage - sequencer will run with minimal data")
		return nil
	}

	// Load a subset of transactions for simulation
	criteria := &DatasetCriteria{
		MaxTransactions: 1000, // Load up to 1000 transactions
		SortBy:          "block",
		SortDesc:        true, // Get most recent first
	}

	dataset, err := mss.storage.ExportDataset(criteria)
	if err != nil {
		return fmt.Errorf("failed to export dataset: %w", err)
	}

	// Convert to sequencer transactions
	mss.transactionQueue = make([]*SequencerTransaction, 0, len(dataset.Transactions))
	for i, tx := range dataset.Transactions {
		seqTx := &SequencerTransaction{
			RealTransactionData: tx,
			SubmittedAt:         time.Now().Add(-time.Duration(len(dataset.Transactions)-i) * time.Second),
			SequenceNumber:      uint64(i),
			Priority:            mss.calculateTransactionPriority(tx),
		}
		mss.transactionQueue = append(mss.transactionQueue, seqTx)
	}

	mss.logger.Info(fmt.Sprintf("Loaded %d transactions for sequencer simulation", len(mss.transactionQueue)))
	return nil
}

// calculateTransactionPriority calculates transaction priority for sequencing
func (mss *MockSequencerService) calculateTransactionPriority(tx *RealTransactionData) int {
	priority := 0

	// Higher gas price = higher priority
	if tx.GasPrice != nil {
		priority += int(tx.GasPrice.Uint64() / 1e9) // Convert to gwei
	}

	// MEV transactions get higher priority
	switch tx.MEVClassification {
	case "potential_arbitrage":
		priority += 100
	case "large_swap":
		priority += 50
	case "high_slippage":
		priority += 25
	}

	// Add randomness for realistic simulation
	priority += rand.Intn(20) - 10

	return priority
}

// blockProductionLoop produces blocks at regular intervals
func (mss *MockSequencerService) blockProductionLoop(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-mss.blockTimer.C:
			if err := mss.produceBlock(); err != nil {
				mss.logger.Error(fmt.Sprintf("Failed to produce block: %v", err))
				mss.incrementErrorCount()
			}
		}
	}
}

// batchProcessingLoop processes transaction batches
func (mss *MockSequencerService) batchProcessingLoop(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-mss.batchTimer.C:
			mss.processBatch()
		}
	}
}

// metricsUpdateLoop updates metrics periodically
func (mss *MockSequencerService) metricsUpdateLoop(ctx context.Context) {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			mss.updateMetrics()
		}
	}
}

// produceBlock creates and broadcasts a new block
func (mss *MockSequencerService) produceBlock() error {
	startTime := time.Now()

	mss.mu.Lock()

	// Get transactions for this block
	blockTxs := mss.selectTransactionsForBlock()

	// Update current block number
	mss.currentBlock++

	// Create block
	block := &SequencerBlock{
		Number:       mss.currentBlock,
		Hash:         mss.generateBlockHash(),
		ParentHash:   mss.generateParentHash(),
		Timestamp:    time.Now(),
		Transactions: blockTxs,
		BatchID:      fmt.Sprintf("batch_%d_%d", mss.currentBlock, time.Now().Unix()),
		GasLimit:     32000000, // Arbitrum block gas limit
	}

	// Calculate block metrics
	var totalGasUsed uint64
	var totalCompressionSize int
	var totalOriginalSize int

	for _, tx := range blockTxs {
		totalGasUsed += tx.GasUsed

		// Simulate compression
		originalSize := len(tx.Data) + 200                                        // Approximate transaction overhead
		compressedSize := int(float64(originalSize) * (0.3 + rand.Float64()*0.4)) // 30-70% compression
		tx.CompressionSize = compressedSize

		totalCompressionSize += compressedSize
		totalOriginalSize += originalSize

		// Set processing time
		tx.ProcessedAt = time.Now()
		tx.ValidationTime = time.Duration(rand.Intn(10)+1) * time.Millisecond
		tx.InclusionDelay = time.Since(tx.SubmittedAt)
	}

	block.GasUsed = totalGasUsed
	if totalOriginalSize > 0 {
		block.CompressionRatio = float64(totalCompressionSize) / float64(totalOriginalSize)
	}
	block.ProcessingTime = time.Since(startTime)

	mss.mu.Unlock()

	// Update metrics
	mss.updateBlockMetrics(block)

	// Broadcast to subscribers
	mss.broadcastBlock(block)

	mss.logger.Debug(fmt.Sprintf("Produced block %d with %d transactions (%.2f%% compression, %v processing time)",
		block.Number, len(block.Transactions), block.CompressionRatio*100, block.ProcessingTime))

	return nil
}

// selectTransactionsForBlock selects transactions for inclusion in a block
func (mss *MockSequencerService) selectTransactionsForBlock() []*SequencerTransaction {
	// Sort transactions by priority
	sort.Slice(mss.transactionQueue, func(i, j int) bool {
		return mss.transactionQueue[i].Priority > mss.transactionQueue[j].Priority
	})

	// Select transactions up to gas limit or batch size
	var selectedTxs []*SequencerTransaction
	var totalGas uint64
	maxGas := uint64(32000000) // Arbitrum block gas limit
	maxTxs := mss.config.BatchSize

	for i, tx := range mss.transactionQueue {
		if len(selectedTxs) >= maxTxs || totalGas+tx.GasUsed > maxGas {
			break
		}

		selectedTxs = append(selectedTxs, tx)
		totalGas += tx.GasUsed

		// Remove from queue
		mss.transactionQueue = append(mss.transactionQueue[:i], mss.transactionQueue[i+1:]...)
	}

	// Replenish queue with new simulated transactions if running low
	if len(mss.transactionQueue) < 10 {
		mss.generateSimulatedTransactions(50)
	}

	return selectedTxs
}

// generateSimulatedTransactions creates simulated transactions to keep the queue populated
func (mss *MockSequencerService) generateSimulatedTransactions(count int) {
	for i := 0; i < count; i++ {
		tx := mss.createSimulatedTransaction()
		mss.transactionQueue = append(mss.transactionQueue, tx)
	}
}

// createSimulatedTransaction creates a realistic simulated transaction
func (mss *MockSequencerService) createSimulatedTransaction() *SequencerTransaction {
	// Create base transaction data
	hash := common.HexToHash(fmt.Sprintf("0x%064x", rand.Uint64()))

	protocols := []string{"UniswapV3", "UniswapV2", "SushiSwap", "Camelot", "1Inch"}
	protocol := protocols[rand.Intn(len(protocols))]

	mevTypes := []string{"regular_swap", "large_swap", "potential_arbitrage", "high_slippage"}
	mevType := mevTypes[rand.Intn(len(mevTypes))]

	// Simulate realistic swap values
	minValue := 100.0
	maxValue := 50000.0
	valueUSD := minValue + rand.Float64()*(maxValue-minValue)

	tx := &SequencerTransaction{
		RealTransactionData: &RealTransactionData{
			Hash:               hash,
			BlockNumber:        mss.currentBlock + 1,
			From:               common.HexToAddress(fmt.Sprintf("0x%040x", rand.Uint64())),
			To:                 &common.Address{},
			GasUsed:            uint64(21000 + rand.Intn(500000)), // 21k to 521k gas
			EstimatedValueUSD:  valueUSD,
			MEVClassification:  mevType,
			SequencerTimestamp: time.Now(),
			ParsedDEX: &arbitrum.DEXTransaction{
				Protocol: protocol,
			},
		},
		SubmittedAt:    time.Now(),
		SequenceNumber: uint64(len(mss.transactionQueue)),
		Priority:       mss.calculatePriorityFromValue(valueUSD, mevType),
	}

	return tx
}

// calculatePriorityFromValue calculates priority based on transaction value and type
func (mss *MockSequencerService) calculatePriorityFromValue(valueUSD float64, mevType string) int {
	priority := int(valueUSD / 1000) // Base priority from value

	switch mevType {
	case "potential_arbitrage":
		priority += 100
	case "large_swap":
		priority += 50
	case "high_slippage":
		priority += 25
	}

	return priority + rand.Intn(20) - 10
}

// processBatch processes a batch of transactions (simulation of mempool processing)
func (mss *MockSequencerService) processBatch() {
	// Simulate adding new transactions to the queue
	newTxCount := rand.Intn(10) + 1 // 1-10 new transactions per batch
	mss.generateSimulatedTransactions(newTxCount)

	// Simulate validation and preprocessing
	for _, tx := range mss.transactionQueue[len(mss.transactionQueue)-newTxCount:] {
		tx.ValidationTime = time.Duration(rand.Intn(5)+1) * time.Millisecond
	}
}

// generateBlockHash generates a realistic block hash
func (mss *MockSequencerService) generateBlockHash() common.Hash {
	return common.HexToHash(fmt.Sprintf("0x%064x", rand.Uint64()))
}

// generateParentHash generates a parent block hash
func (mss *MockSequencerService) generateParentHash() common.Hash {
	return common.HexToHash(fmt.Sprintf("0x%064x", rand.Uint64()))
}

// updateBlockMetrics updates metrics after block production
func (mss *MockSequencerService) updateBlockMetrics(block *SequencerBlock) {
	mss.metrics.mu.Lock()
	defer mss.metrics.mu.Unlock()

	mss.metrics.BlocksProduced++
	mss.metrics.TransactionsProcessed += uint64(len(block.Transactions))
	mss.metrics.TotalProcessingTime += block.ProcessingTime
	mss.metrics.LastBlockTime = block.Timestamp
	mss.metrics.QueueSize = len(mss.transactionQueue)

	// Count DEX and MEV transactions
	for _, tx := range block.Transactions {
		if tx.ParsedDEX != nil {
			mss.metrics.DEXTransactionsFound++
		}
		if tx.MEVClassification != "regular_swap" {
			mss.metrics.MEVTransactionsFound++
		}
	}

	// Update averages
	if mss.metrics.BlocksProduced > 0 {
		mss.metrics.AverageTxPerBlock = float64(mss.metrics.TransactionsProcessed) / float64(mss.metrics.BlocksProduced)
		mss.metrics.AverageBlockTime = mss.metrics.TotalProcessingTime / time.Duration(mss.metrics.BlocksProduced)
	}

	// Update compression ratio
	mss.metrics.AverageCompressionRatio = (mss.metrics.AverageCompressionRatio*float64(mss.metrics.BlocksProduced-1) + block.CompressionRatio) / float64(mss.metrics.BlocksProduced)
}

// updateMetrics updates real-time metrics
func (mss *MockSequencerService) updateMetrics() {
	mss.metrics.mu.Lock()
	defer mss.metrics.mu.Unlock()

	mss.metrics.QueueSize = len(mss.transactionQueue)
	mss.metrics.SubscriberCount = len(mss.subscribers)
}

// incrementErrorCount increments the error count
func (mss *MockSequencerService) incrementErrorCount() {
	mss.metrics.mu.Lock()
	defer mss.metrics.mu.Unlock()
	mss.metrics.ErrorCount++
}

// broadcastBlock broadcasts a block to all subscribers
func (mss *MockSequencerService) broadcastBlock(block *SequencerBlock) {
	mss.mu.RLock()
	defer mss.mu.RUnlock()

	for id, ch := range mss.subscribers {
		select {
		case ch <- block:
			// Block sent successfully
		default:
			// Channel is full or closed, remove subscriber
			mss.logger.Warn(fmt.Sprintf("Removing unresponsive subscriber: %s", id))
			close(ch)
			delete(mss.subscribers, id)
		}
	}
}

// Subscribe subscribes to block updates
func (mss *MockSequencerService) Subscribe(id string) <-chan *SequencerBlock {
	mss.mu.Lock()
	defer mss.mu.Unlock()

	ch := make(chan *SequencerBlock, 10) // Buffer for 10 blocks
	mss.subscribers[id] = ch

	mss.logger.Debug(fmt.Sprintf("New subscriber: %s", id))
	return ch
}

// Unsubscribe unsubscribes from block updates
func (mss *MockSequencerService) Unsubscribe(id string) {
	mss.mu.Lock()
	defer mss.mu.Unlock()

	if ch, exists := mss.subscribers[id]; exists {
		close(ch)
		delete(mss.subscribers, id)
		mss.logger.Debug(fmt.Sprintf("Unsubscribed: %s", id))
	}
}

// GetMetrics returns current sequencer metrics
func (mss *MockSequencerService) GetMetrics() *SequencerMetrics {
	mss.metrics.mu.RLock()
	defer mss.metrics.mu.RUnlock()

	// Return a copy
	metricsCopy := *mss.metrics
	return &metricsCopy
}

// GetCurrentBlock returns the current block number
func (mss *MockSequencerService) GetCurrentBlock() uint64 {
	mss.mu.RLock()
	defer mss.mu.RUnlock()
	return mss.currentBlock
}

// GetQueueSize returns the current transaction queue size
func (mss *MockSequencerService) GetQueueSize() int {
	mss.mu.RLock()
	defer mss.mu.RUnlock()
	return len(mss.transactionQueue)
}

// SimulateMEVBurst simulates a burst of MEV activity
func (mss *MockSequencerService) SimulateMEVBurst(txCount int) {
	mss.mu.Lock()
	defer mss.mu.Unlock()

	mss.logger.Info(fmt.Sprintf("Simulating MEV burst with %d transactions", txCount))

	for i := 0; i < txCount; i++ {
		tx := mss.createSimulatedTransaction()

		// Make it an MEV transaction
		mevTypes := []string{"potential_arbitrage", "large_swap", "high_slippage"}
		tx.MEVClassification = mevTypes[rand.Intn(len(mevTypes))]
		tx.EstimatedValueUSD = 10000 + rand.Float64()*90000 // $10k-$100k
		tx.Priority = mss.calculatePriorityFromValue(tx.EstimatedValueUSD, tx.MEVClassification)

		mss.transactionQueue = append(mss.transactionQueue, tx)
	}
}

// ExportMetrics exports sequencer metrics to a file
func (mss *MockSequencerService) ExportMetrics(filename string) error {
	metrics := mss.GetMetrics()

	data, err := json.MarshalIndent(metrics, "", "  ")
	if err != nil {
		return fmt.Errorf("failed to marshal metrics: %w", err)
	}

	if err := mss.storage.saveToDataDir(filename, data); err != nil {
		return fmt.Errorf("failed to save metrics: %w", err)
	}

	mss.logger.Info(fmt.Sprintf("Exported sequencer metrics to %s", filename))
	return nil
}

// saveToDataDir is a helper method for storage
func (ts *TransactionStorage) saveToDataDir(filename string, data []byte) error {
	filePath := filepath.Join(ts.dataDir, filename)
	return os.WriteFile(filePath, data, 0644)
}

// GetRecentBlocks returns recently produced blocks
func (mss *MockSequencerService) GetRecentBlocks(count int) []*SequencerBlock {
	// This would maintain a history of recent blocks in a real implementation
	// For now, return empty slice as this is primarily for testing
	return []*SequencerBlock{}
}