mev-beta/orig/pkg/scanner/concurrent_test.go

//go:build legacy_scanner
// +build legacy_scanner

package scanner

import (
	"math/big"
	"testing"
	"time"

	"github.com/ethereum/go-ethereum/common"
	"github.com/holiman/uint256"
	"github.com/stretchr/testify/assert"

	"github.com/fraktal/mev-beta/internal/config"
	"github.com/fraktal/mev-beta/internal/logger"
	"github.com/fraktal/mev-beta/pkg/contracts"
	"github.com/fraktal/mev-beta/pkg/database"
	"github.com/fraktal/mev-beta/pkg/events"
)

func TestNewMarketScanner(t *testing.T) {
	// Create test config
	cfg := &config.BotConfig{
		MaxWorkers: 5,
		RPCTimeout: 30,
	}

	// Create test logger
	logger := logger.New("info", "text", "")

	// Create mock contract executor and database
	var contractExecutor *contracts.ContractExecutor // nil for testing
	var db *database.Database                        // nil for testing

	// Create market scanner
	scanner := NewMarketScanner(cfg, logger, contractExecutor, db)

	// Verify scanner was created correctly
	assert.NotNil(t, scanner)
	assert.Equal(t, cfg, scanner.config)
	assert.Equal(t, logger, scanner.logger)
	assert.NotNil(t, scanner.workerPool)
	assert.NotNil(t, scanner.workers)
	assert.NotNil(t, scanner.cache)
	assert.NotNil(t, scanner.cacheTTL)
	assert.Equal(t, time.Duration(cfg.RPCTimeout)*time.Second, scanner.cacheTTL)
	assert.Equal(t, cfg.MaxWorkers, len(scanner.workers))
}

func TestEventTypeString(t *testing.T) {
	// Test all event types
	assert.Equal(t, "Unknown", events.Unknown.String())
	assert.Equal(t, "Swap", events.Swap.String())
	assert.Equal(t, "AddLiquidity", events.AddLiquidity.String())
	assert.Equal(t, "RemoveLiquidity", events.RemoveLiquidity.String())
	assert.Equal(t, "NewPool", events.NewPool.String())
}

func TestIsSignificantMovement(t *testing.T) {
	// Create market scanner
	cfg := &config.BotConfig{
		MinProfitThreshold: 10.0,
	}
	logger := logger.New("info", "text", "")
	scanner := NewMarketScanner(cfg, logger)

	// Test significant movement
	movement := &PriceMovement{
		PriceImpact: 15.0, // Above threshold
	}
	assert.True(t, scanner.isSignificantMovement(movement, cfg.MinProfitThreshold))

	// Test insignificant movement
	movement = &PriceMovement{
		PriceImpact: 5.0, // Below threshold
	}
	assert.False(t, scanner.isSignificantMovement(movement, cfg.MinProfitThreshold))
}

func TestCalculatePriceMovement(t *testing.T) {
	// Create market scanner
	cfg := &config.BotConfig{}
	logger := logger.New("info", "text", "")
	scanner := NewMarketScanner(cfg, logger)

	// Create test event
	event := events.Event{
		Token0:    common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
		Token1:    common.HexToAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
		Amount0:   big.NewInt(1000000000),         // 1000 tokens
		Amount1:   big.NewInt(500000000000000000), // 0.5 ETH
		Tick:      200000,
		Timestamp: uint64(time.Now().Unix()),
	}

	// Create test pool data
	poolData := &CachedData{
		SqrtPriceX96: uint256.NewInt(2505414483750470000),
	}

	// Calculate price movement
	priceMovement, err := scanner.calculatePriceMovement(event, poolData)

	// Verify results
	assert.NoError(t, err)
	assert.NotNil(t, priceMovement)
	assert.Equal(t, event.Token0.Hex(), priceMovement.Token0)
	assert.Equal(t, event.Token1.Hex(), priceMovement.Token1)
	assert.Equal(t, event.Tick, priceMovement.TickBefore)
	// Note: We're not strictly comparing timestamps since the implementation uses time.Now()
	assert.NotNil(t, priceMovement.Timestamp)
	assert.NotNil(t, priceMovement.PriceBefore)
	assert.NotNil(t, priceMovement.AmountIn)
	assert.NotNil(t, priceMovement.AmountOut)
}

func TestFindArbitrageOpportunities(t *testing.T) {
	// Create market scanner
	cfg := &config.BotConfig{}
	logger := logger.New("info", "text", "")
	scanner := NewMarketScanner(cfg, logger)

	// Create test event
	event := events.Event{
		PoolAddress: common.HexToAddress("0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640"),
		Token0:      common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
		Token1:      common.HexToAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
		Protocol:    "UniswapV3",
		Amount0:     big.NewInt(1000000000),         // 1000 tokens
		Amount1:     big.NewInt(500000000000000000), // 0.5 ETH
	}

	// Create test price movement
	movement := &PriceMovement{
		Token0:      event.Token0.Hex(),
		Token1:      event.Token1.Hex(),
		Pool:        event.PoolAddress.Hex(),
		Protocol:    event.Protocol,
		PriceImpact: 5.0,
		Timestamp:   time.Now(),
		PriceBefore: big.NewFloat(2000.0), // Mock price
	}

	// Find arbitrage opportunities (should return mock opportunities)
	opportunities := scanner.findArbitrageOpportunities(event, movement)

	// Verify results
	assert.NotNil(t, opportunities)
	// Note: The number of opportunities depends on the mock data and may vary
	// Just verify that the function doesn't panic and returns a slice
	assert.NotNil(t, opportunities)
}

func TestGetPoolDataCacheHit(t *testing.T) {
	// Create market scanner
	cfg := &config.BotConfig{
		RPCTimeout: 30,
	}
	logger := logger.New("info", "text", "")
	scanner := NewMarketScanner(cfg, logger)

	// Add pool data to cache
	poolAddress := "0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640"
	poolData := &CachedData{
		Address:      common.HexToAddress(poolAddress),
		Token0:       common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
		Token1:       common.HexToAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
		Fee:          3000,
		Liquidity:    uint256.NewInt(1000000000000000000),
		SqrtPriceX96: uint256.NewInt(2505414483750470000),
		Tick:         200000,
		TickSpacing:  60,
		LastUpdated:  time.Now(),
	}
	scanner.cacheMutex.Lock()
	scanner.cache["pool_"+poolAddress] = poolData
	scanner.cacheMutex.Unlock()

	// Get pool data (should be cache hit)
	result, err := scanner.getPoolData(poolAddress)

	// Verify results
	assert.NoError(t, err)
	assert.Equal(t, poolData, result)
}

func TestUpdatePoolData(t *testing.T) {
	// Create market scanner
	cfg := &config.BotConfig{}
	logger := logger.New("info", "text", "")
	scanner := NewMarketScanner(cfg, logger)

	// Create test event
	event := events.Event{
		PoolAddress:  common.HexToAddress("0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640"),
		Token0:       common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
		Token1:       common.HexToAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
		Liquidity:    uint256.NewInt(1000000000000000000),
		SqrtPriceX96: uint256.NewInt(2505414483750470000),
		Tick:         200000,
		Timestamp:    uint64(time.Now().Unix()),
	}

	// Update pool data
	scanner.updatePoolData(event)

	// Verify the pool data was updated
	scanner.cacheMutex.RLock()
	poolData, exists := scanner.cache["pool_"+event.PoolAddress.Hex()]
	scanner.cacheMutex.RUnlock()

	assert.True(t, exists)
	assert.NotNil(t, poolData)
	assert.Equal(t, event.PoolAddress, poolData.Address)
	assert.Equal(t, event.Token0, poolData.Token0)
	assert.Equal(t, event.Token1, poolData.Token1)
	assert.Equal(t, event.Liquidity, poolData.Liquidity)
	assert.Equal(t, event.SqrtPriceX96, poolData.SqrtPriceX96)
	assert.Equal(t, event.Tick, poolData.Tick)
}

// RACE CONDITION FIX TEST: Test concurrent worker processing without race conditions
func TestConcurrentWorkerProcessingRaceDetection(t *testing.T) {
	// Create test config with multiple workers
	cfg := &config.BotConfig{
		MaxWorkers: 10,
		RPCTimeout: 30,
	}

	// Create test logger
	logger := logger.New("info", "text", "")

	// Mock database
	db, err := database.NewInMemoryDatabase()
	assert.NoError(t, err)

	// Mock contracts registry
	contractsRegistry := &contracts.ContractsRegistry{}

	// Create scanner
	scanner := NewMarketScanner(cfg, logger)
	scanner.db = db
	scanner.contracts = contractsRegistry

	// Create multiple test events to simulate concurrent processing
	events := make([]events.Event, 100)
	for i := 0; i < 100; i++ {
		events[i] = events.Event{
			Type:        events.Swap,
			PoolAddress: common.BigToAddress(big.NewInt(int64(i))),
			Token0:      common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
			Token1:      common.HexToAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
			Liquidity:   uint256.NewInt(1000000000000000000),
			Timestamp:   uint64(time.Now().Unix()),
		}
	}

	// Submit all events concurrently
	start := time.Now()
	for _, event := range events {
		scanner.SubmitEvent(event)
	}

	// Wait for all processing to complete
	scanner.WaitGroup().Wait()
	duration := time.Since(start)

	// Test should complete without hanging (indicates no race condition)
	assert.Less(t, duration, 10*time.Second, "Processing took too long, possible race condition")

	t.Logf("Successfully processed %d events in %v", len(events), duration)
}

// RACE CONDITION FIX TEST: Stress test with high concurrency
func TestHighConcurrencyStressTest(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping stress test in short mode")
	}

	// Create test config with many workers
	cfg := &config.BotConfig{
		MaxWorkers: 50,
		RPCTimeout: 30,
	}

	// Create test logger
	logger := logger.New("info", "text", "")

	// Mock database
	db, err := database.NewInMemoryDatabase()
	assert.NoError(t, err)

	// Mock contracts registry
	contractsRegistry := &contracts.ContractsRegistry{}

	// Create scanner
	scanner := NewMarketScanner(cfg, logger)
	scanner.db = db
	scanner.contracts = contractsRegistry

	// Create many test events
	numEvents := 1000
	events := make([]events.Event, numEvents)
	for i := 0; i < numEvents; i++ {
		events[i] = events.Event{
			Type:        events.Swap,
			PoolAddress: common.BigToAddress(big.NewInt(int64(i))),
			Token0:      common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
			Token1:      common.HexToAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
			Liquidity:   uint256.NewInt(uint64(1000000000000000000 + i)),
			Timestamp:   uint64(time.Now().Unix()),
		}
	}

	// Submit all events rapidly
	start := time.Now()
	for _, event := range events {
		scanner.SubmitEvent(event)
	}

	// Wait for all processing to complete
	scanner.WaitGroup().Wait()
	duration := time.Since(start)

	// Test should complete without hanging or panicking
	assert.Less(t, duration, 30*time.Second, "High concurrency processing took too long")

	t.Logf("Successfully processed %d events with %d workers in %v",
		numEvents, cfg.MaxWorkers, duration)
}