mev-beta/pkg/scanner/concurrent.go

package scanner

import (
	"fmt"
	"math/big"
	"sync"
	"time"

	"github.com/fraktal/mev-beta/internal/config"
	"github.com/fraktal/mev-beta/internal/logger"
	"github.com/fraktal/mev-beta/pkg/circuit"
	"github.com/fraktal/mev-beta/pkg/events"
	"github.com/fraktal/mev-beta/pkg/trading"
	"github.com/fraktal/mev-beta/pkg/uniswap"
	"github.com/ethereum/go-ethereum/common"
	"github.com/holiman/uint256"
	"golang.org/x/sync/singleflight"
)

// MarketScanner scans markets for price movement opportunities with concurrency
type MarketScanner struct {
	config            *config.BotConfig
	logger            *logger.Logger
	workerPool        chan chan events.Event
	workers           []*EventWorker
	wg                sync.WaitGroup
	cacheGroup        singleflight.Group
	cache             map[string]*CachedData
	cacheMutex        sync.RWMutex
	cacheTTL          time.Duration
	slippageProtector *trading.SlippageProtection
	circuitBreaker    *circuit.CircuitBreaker
}

// EventWorker represents a worker that processes event details
type EventWorker struct {
	ID         int
	WorkerPool chan chan events.Event
	JobChannel chan events.Event
	QuitChan   chan bool
	scanner    *MarketScanner
}

// NewMarketScanner creates a new market scanner with concurrency support
func NewMarketScanner(cfg *config.BotConfig, logger *logger.Logger) *MarketScanner {
	scanner := &MarketScanner{
		config:            cfg,
		logger:            logger,
		workerPool:        make(chan chan events.Event, cfg.MaxWorkers),
		workers:           make([]*EventWorker, 0, cfg.MaxWorkers),
		cache:             make(map[string]*CachedData),
		cacheTTL:          time.Duration(cfg.RPCTimeout) * time.Second,
		slippageProtector: trading.NewSlippageProtection(logger),
		circuitBreaker:    circuit.NewCircuitBreaker(&circuit.Config{
			Logger:           logger,
			Name:            "market_scanner",
			MaxFailures:     10,
			ResetTimeout:    time.Minute * 5,
			MaxRequests:     3,
			SuccessThreshold: 2,
		}),
	}

	// Create workers
	for i := 0; i < cfg.MaxWorkers; i++ {
		worker := NewEventWorker(i, scanner.workerPool, scanner)
		scanner.workers = append(scanner.workers, worker)
		worker.Start()
	}

	// Start cache cleanup routine
	go scanner.cleanupCache()

	return scanner
}

// NewEventWorker creates a new event worker
func NewEventWorker(id int, workerPool chan chan events.Event, scanner *MarketScanner) *EventWorker {
	return &EventWorker{
		ID:         id,
		WorkerPool: workerPool,
		JobChannel: make(chan events.Event),
		QuitChan:   make(chan bool),
		scanner:    scanner,
	}
}

// Start begins the worker
func (w *EventWorker) Start() {
	go func() {
		for {
			// Register the worker in the worker pool
			w.WorkerPool <- w.JobChannel

			select {
			case job := <-w.JobChannel:
				// Process the job
				w.Process(job)
			case <-w.QuitChan:
				// Stop the worker
				return
			}
		}
	}()
}

// Stop terminates the worker
func (w *EventWorker) Stop() {
	go func() {
		w.QuitChan <- true
	}()
}

// Process handles an event detail
func (w *EventWorker) Process(event events.Event) {
	// Analyze the event in a separate goroutine to maintain throughput
	go func() {
		defer w.scanner.wg.Done()

		// Log the processing
		w.scanner.logger.Debug(fmt.Sprintf("Worker %d processing %s event in pool %s from protocol %s",
			w.ID, event.Type.String(), event.PoolAddress, event.Protocol))

		// Analyze based on event type
		switch event.Type {
		case events.Swap:
			w.scanner.analyzeSwapEvent(event)
		case events.AddLiquidity:
			w.scanner.analyzeLiquidityEvent(event, true)
		case events.RemoveLiquidity:
			w.scanner.analyzeLiquidityEvent(event, false)
		case events.NewPool:
			w.scanner.analyzeNewPoolEvent(event)
		default:
			w.scanner.logger.Debug(fmt.Sprintf("Worker %d received unknown event type: %d", w.ID, event.Type))
		}
	}()
}

// SubmitEvent submits an event for processing by the worker pool
func (s *MarketScanner) SubmitEvent(event events.Event) {
	s.wg.Add(1)

	// Get an available worker job channel
	jobChannel := <-s.workerPool

	// Send the job to the worker
	jobChannel <- event
}

// analyzeSwapEvent analyzes a swap event for arbitrage opportunities
func (s *MarketScanner) analyzeSwapEvent(event events.Event) {
	s.logger.Debug(fmt.Sprintf("Analyzing swap event in pool %s", event.PoolAddress))

	// Get pool data with caching
	poolData, err := s.getPoolData(event.PoolAddress.Hex())
	if err != nil {
		s.logger.Error(fmt.Sprintf("Error getting pool data for %s: %v", event.PoolAddress, err))
		return
	}

	// Calculate price impact
	priceMovement, err := s.calculatePriceMovement(event, poolData)
	if err != nil {
		s.logger.Error(fmt.Sprintf("Error calculating price movement for pool %s: %v", event.PoolAddress, err))
		return
	}

	// Check if the movement is significant
	if s.isSignificantMovement(priceMovement, s.config.MinProfitThreshold) {
		s.logger.Info(fmt.Sprintf("Significant price movement detected in pool %s: %+v", event.PoolAddress, priceMovement))

		// Look for arbitrage opportunities
		opportunities := s.findArbitrageOpportunities(event, priceMovement)
		if len(opportunities) > 0 {
			s.logger.Info(fmt.Sprintf("Found %d arbitrage opportunities for pool %s", len(opportunities), event.PoolAddress))
			for _, opp := range opportunities {
				s.logger.Info(fmt.Sprintf("Arbitrage opportunity: %+v", opp))
			}
		}
	} else {
		s.logger.Debug(fmt.Sprintf("Price movement in pool %s is not significant: %f", event.PoolAddress, priceMovement.PriceImpact))
	}
}

// analyzeLiquidityEvent analyzes liquidity events (add/remove)
func (s *MarketScanner) analyzeLiquidityEvent(event events.Event, isAdd bool) {
	action := "adding"
	if !isAdd {
		action = "removing"
	}
	s.logger.Debug(fmt.Sprintf("Analyzing liquidity event (%s) in pool %s", action, event.PoolAddress))

	// Update cached pool data
	s.updatePoolData(event)

	s.logger.Info(fmt.Sprintf("Liquidity %s event processed for pool %s", action, event.PoolAddress))
}

// analyzeNewPoolEvent analyzes new pool creation events
func (s *MarketScanner) analyzeNewPoolEvent(event events.Event) {
	s.logger.Info(fmt.Sprintf("New pool created: %s (protocol: %s)", event.PoolAddress, event.Protocol))

	// Add to known pools
	// In a real implementation, you would want to fetch and cache the pool data
	s.logger.Debug(fmt.Sprintf("Added new pool %s to monitoring", event.PoolAddress))
}

// calculatePriceMovement calculates the price movement from a swap event
func (s *MarketScanner) calculatePriceMovement(event events.Event, poolData *CachedData) (*PriceMovement, error) {
	// Calculate the price before the swap using Uniswap V3 math
	priceBefore := uniswap.SqrtPriceX96ToPrice(poolData.SqrtPriceX96.ToBig())
	
	// For a more accurate calculation, we would need to:
	// 1. Calculate the price after the swap using Uniswap V3 math
	// 2. Account for liquidity changes
	// 3. Consider the tick spacing and fee
	
	priceMovement := &PriceMovement{
		Token0:      event.Token0.Hex(),
		Token1:      event.Token1.Hex(),
		Pool:        event.PoolAddress.Hex(),
		Protocol:    event.Protocol,
		AmountIn:    new(big.Int).Set(event.Amount0),
		AmountOut:   new(big.Int).Set(event.Amount1),
		PriceBefore: priceBefore,
		TickBefore:  event.Tick,
		Timestamp:   time.Now(), // In a real implementation, use the actual event timestamp
	}
	
	// Calculate price impact using a more realistic approach
	// For Uniswap V3, price impact is roughly amountIn / liquidity
	if event.Liquidity != nil && event.Liquidity.Sign() > 0 && event.Amount0 != nil && event.Amount0.Sign() > 0 {
		liquidityFloat := new(big.Float).SetInt(event.Liquidity.ToBig())
		amountInFloat := new(big.Float).SetInt(event.Amount0)
		
		// Price impact ≈ amountIn / liquidity
		priceImpact := new(big.Float).Quo(amountInFloat, liquidityFloat)
		priceImpactFloat, _ := priceImpact.Float64()
		priceMovement.PriceImpact = priceImpactFloat
	} else if priceMovement.AmountIn.Cmp(big.NewInt(0)) > 0 {
		// Fallback calculation
		impact := new(big.Float).Quo(
			new(big.Float).SetInt(priceMovement.AmountOut),
			new(big.Float).SetInt(priceMovement.AmountIn),
		)
		priceImpact, _ := impact.Float64()
		priceMovement.PriceImpact = priceImpact
	}
	
	return priceMovement, nil
}

// isSignificantMovement determines if a price movement is significant enough to exploit
func (s *MarketScanner) isSignificantMovement(movement *PriceMovement, threshold float64) bool {
	// Check if the price impact is above our threshold
	if movement.PriceImpact > threshold {
		return true
	}
	
	// Also check if the absolute amount is significant
	if movement.AmountIn != nil && movement.AmountIn.Cmp(big.NewInt(1000000000000000000)) > 0 { // 1 ETH
		return true
	}
	
	// For smaller amounts, we need a higher price impact to be significant
	if movement.AmountIn != nil && movement.AmountIn.Cmp(big.NewInt(100000000000000000)) > 0 { // 0.1 ETH
		return movement.PriceImpact > threshold/2
	}
	
	return false
}

// findRelatedPools finds pools that trade the same token pair
func (s *MarketScanner) findRelatedPools(token0, token1 common.Address) []*CachedData {
	s.logger.Debug(fmt.Sprintf("Finding related pools for token pair %s-%s", token0.Hex(), token1.Hex()))
	
	relatedPools := make([]*CachedData, 0)
	
	// In a real implementation, this would query a pool registry or
	// search through known pools for pools with the same token pair
	// For now, we'll return some mock data
	
	// Check if we have cached data for common pools
	commonPools := []string{
		"0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640", // USDC/WETH Uniswap V3 0.05%
		"0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc", // USDC/WETH Uniswap V2 0.3%
	}
	
	for _, poolAddr := range commonPools {
		poolData, err := s.getPoolData(poolAddr)
		if err != nil {
			s.logger.Debug(fmt.Sprintf("No data for pool %s: %v", poolAddr, err))
			continue
		}
		
		// Check if this pool trades the same token pair (in either direction)
		if (poolData.Token0 == token0 && poolData.Token1 == token1) ||
		   (poolData.Token0 == token1 && poolData.Token1 == token0) {
			relatedPools = append(relatedPools, poolData)
		}
	}
	
	s.logger.Debug(fmt.Sprintf("Found %d related pools", len(relatedPools)))
	return relatedPools
}

// estimateProfit estimates the potential profit from an arbitrage opportunity
func (s *MarketScanner) estimateProfit(event events.Event, pool *CachedData, priceDiff float64) *big.Int {
	// This is a simplified profit estimation
	// In practice, this would involve complex calculations including:
	// - Precise Uniswap V3 math for swap calculations
	// - Gas cost estimation
	// - Slippage calculations
	// - Path optimization
	
	// For now, we'll use a simplified calculation
	amountIn := new(big.Int).Set(event.Amount0)
	priceDiffInt := big.NewInt(int64(priceDiff * 1000000)) // Scale for integer math
	
	// Estimated profit = amount * price difference
	profit := new(big.Int).Mul(amountIn, priceDiffInt)
	profit = profit.Div(profit, big.NewInt(1000000))
	
	// Subtract estimated gas costs
	gasCost := big.NewInt(300000) // Rough estimate
	profit = profit.Sub(profit, gasCost)
	
	// Ensure profit is positive
	if profit.Sign() <= 0 {
		return big.NewInt(0)
	}
	
	return profit
}

// findTriangularArbitrageOpportunities looks for triangular arbitrage opportunities
func (s *MarketScanner) findTriangularArbitrageOpportunities(event events.Event) []ArbitrageOpportunity {
	s.logger.Debug(fmt.Sprintf("Searching for triangular arbitrage opportunities involving pool %s", event.PoolAddress.Hex()))
	
	opportunities := make([]ArbitrageOpportunity, 0)
	
	// This would implement logic to find triangular arbitrage paths like:
	// TokenA -> TokenB -> TokenC -> TokenA
	// where the end balance of TokenA is greater than the starting balance
	
	// For now, we'll return an empty slice
	// A full implementation would:
	// 1. Identify common triangular paths (e.g., USDC -> WETH -> WBTC -> USDC)
	// 2. Calculate the output of each leg of the trade
	// 3. Account for all fees and slippage
	// 4. Compare the final amount with the initial amount
	
	return opportunities
}

// findArbitrageOpportunities looks for arbitrage opportunities based on price movements
func (s *MarketScanner) findArbitrageOpportunities(event events.Event, movement *PriceMovement) []ArbitrageOpportunity {
	s.logger.Debug(fmt.Sprintf("Searching for arbitrage opportunities for pool %s", event.PoolAddress))

	opportunities := make([]ArbitrageOpportunity, 0)

	// Get related pools for the same token pair
	relatedPools := s.findRelatedPools(event.Token0, event.Token1)
	
	// If we have related pools, compare prices
	if len(relatedPools) > 0 {
		// Get the current price in this pool
		currentPrice := movement.PriceBefore
		
		// Compare with prices in related pools
		for _, pool := range relatedPools {
			// Skip the same pool
			if pool.Address == event.PoolAddress {
				continue
			}
			
			// Get pool data
			poolData, err := s.getPoolData(pool.Address.Hex())
			if err != nil {
				s.logger.Error(fmt.Sprintf("Error getting pool data for related pool %s: %v", pool.Address.Hex(), err))
				continue
			}
			
			// Check if poolData.SqrtPriceX96 is nil to prevent panic
			if poolData.SqrtPriceX96 == nil {
				s.logger.Error(fmt.Sprintf("Pool data for %s has nil SqrtPriceX96", pool.Address.Hex()))
				continue
			}
			
			// Calculate price in the related pool
			relatedPrice := uniswap.SqrtPriceX96ToPrice(poolData.SqrtPriceX96.ToBig())
			
			// Check if currentPrice or relatedPrice is nil to prevent panic
			if currentPrice == nil || relatedPrice == nil {
				s.logger.Error(fmt.Sprintf("Nil price detected for pool comparison"))
				continue
			}
			
			// Calculate price difference
			priceDiff := new(big.Float).Sub(currentPrice, relatedPrice)
			priceDiffRatio := new(big.Float).Quo(priceDiff, relatedPrice)
			
			// If there's a significant price difference, we might have an arbitrage opportunity
			priceDiffFloat, _ := priceDiffRatio.Float64()
			if priceDiffFloat > 0.005 { // 0.5% threshold
				// Estimate potential profit
				estimatedProfit := s.estimateProfit(event, pool, priceDiffFloat)
				
				if estimatedProfit != nil && estimatedProfit.Sign() > 0 {
					opp := ArbitrageOpportunity{
						Path:        []string{event.Token0.Hex(), event.Token1.Hex()},
						Pools:       []string{event.PoolAddress.Hex(), pool.Address.Hex()},
						Profit:      estimatedProfit,
						GasEstimate: big.NewInt(300000), // Estimated gas cost
						ROI:         priceDiffFloat * 100, // Convert to percentage
						Protocol:    fmt.Sprintf("%s->%s", event.Protocol, pool.Protocol),
					}
					opportunities = append(opportunities, opp)
					s.logger.Info(fmt.Sprintf("Found arbitrage opportunity: %+v", opp))
				}
			}
		}
	}
	
	// Also look for triangular arbitrage opportunities
	triangularOpps := s.findTriangularArbitrageOpportunities(event)
	opportunities = append(opportunities, triangularOpps...)

	return opportunities
}

// Stop stops the market scanner and all workers
func (s *MarketScanner) Stop() {
	// Stop all workers
	for _, worker := range s.workers {
		worker.Stop()
	}

	// Wait for all jobs to complete
	s.wg.Wait()
}

// ArbitrageOpportunity represents a potential arbitrage opportunity
type ArbitrageOpportunity struct {
	Path        []string // Token path for the arbitrage
	Pools       []string // Pools involved in the arbitrage
	Profit      *big.Int // Estimated profit in wei
	GasEstimate *big.Int // Estimated gas cost
	ROI         float64  // Return on investment percentage
	Protocol    string   // DEX protocol
}

// PriceMovement represents a potential price movement
type PriceMovement struct {
	Token0      string     // Token address
	Token1      string     // Token address
	Pool        string     // Pool address
	Protocol    string     // DEX protocol
	AmountIn    *big.Int   // Amount of token being swapped in
	AmountOut   *big.Int   // Amount of token being swapped out
	PriceBefore *big.Float // Price before the swap
	PriceAfter  *big.Float // Price after the swap (to be calculated)
	PriceImpact float64    // Calculated price impact
	TickBefore  int        // Tick before the swap
	TickAfter   int        // Tick after the swap (to be calculated)
	Timestamp   time.Time  // Event timestamp
}

// CachedData represents cached pool data
type CachedData struct {
	Address       common.Address
	Token0        common.Address
	Token1        common.Address
	Fee           int64
	Liquidity     *uint256.Int
	SqrtPriceX96  *uint256.Int
	Tick          int
	TickSpacing   int
	LastUpdated   time.Time
	Protocol      string
}

// getPoolData retrieves pool data with caching
func (s *MarketScanner) getPoolData(poolAddress string) (*CachedData, error) {
	// Check cache first
	cacheKey := fmt.Sprintf("pool_%s", poolAddress)

	s.cacheMutex.RLock()
	if data, exists := s.cache[cacheKey]; exists && time.Since(data.LastUpdated) < s.cacheTTL {
		s.cacheMutex.RUnlock()
		s.logger.Debug(fmt.Sprintf("Cache hit for pool %s", poolAddress))
		return data, nil
	}
	s.cacheMutex.RUnlock()

	// Use singleflight to prevent duplicate requests
	result, err, _ := s.cacheGroup.Do(cacheKey, func() (interface{}, error) {
		return s.fetchPoolData(poolAddress)
	})

	if err != nil {
		return nil, err
	}

	poolData := result.(*CachedData)

	// Update cache
	s.cacheMutex.Lock()
	s.cache[cacheKey] = poolData
	s.cacheMutex.Unlock()

	s.logger.Debug(fmt.Sprintf("Fetched and cached pool data for %s", poolAddress))
	return poolData, nil
}

// fetchPoolData fetches pool data from the blockchain
func (s *MarketScanner) fetchPoolData(poolAddress string) (*CachedData, error) {
	s.logger.Debug(fmt.Sprintf("Fetching pool data for %s", poolAddress))

	// This is a simplified implementation
	// In practice, you would interact with the Ethereum blockchain to get real data
	address := common.HexToAddress(poolAddress)

	// For now, we'll return mock data
	pool := &CachedData{
		Address:      address,
		Token0:       common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"), // USDC
		Token1:       common.HexToAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"), // WETH
		Fee:          3000, // 0.3%
		Liquidity:    uint256.NewInt(1000000000000000000), // 1 ETH equivalent
		SqrtPriceX96: uint256.NewInt(2505414483750470000), // Mock sqrt price
		Tick:         200000, // Mock tick
		TickSpacing:  60,    // Tick spacing for 0.3% fee
		Protocol:     "UniswapV3", // Mock protocol
		LastUpdated:  time.Now(),
	}

	s.logger.Debug(fmt.Sprintf("Fetched pool data for %s", poolAddress))
	return pool, nil
}

// updatePoolData updates cached pool data
func (s *MarketScanner) updatePoolData(event events.Event) {
	cacheKey := fmt.Sprintf("pool_%s", event.PoolAddress.Hex())

	s.cacheMutex.Lock()
	defer s.cacheMutex.Unlock()

	// Update existing cache entry or create new one
	data := &CachedData{
		Address:      event.PoolAddress,
		Token0:       event.Token0,
		Token1:       event.Token1,
		Liquidity:    event.Liquidity,
		SqrtPriceX96: event.SqrtPriceX96,
		Tick:         event.Tick,
		Protocol:     event.Protocol, // Add protocol information
		LastUpdated:  time.Now(),
	}

	s.cache[cacheKey] = data
	s.logger.Debug(fmt.Sprintf("Updated cache for pool %s", event.PoolAddress.Hex()))
}

// cleanupCache removes expired cache entries
func (s *MarketScanner) cleanupCache() {
	ticker := time.NewTicker(10 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			s.cacheMutex.Lock()
			for key, data := range s.cache {
				if time.Since(data.LastUpdated) > s.cacheTTL {
					delete(s.cache, key)
					s.logger.Debug(fmt.Sprintf("Removed expired cache entry: %s", key))
				}
			}
			s.cacheMutex.Unlock()
		}
	}
}