mev-beta/pkg/market/manager.go

package market

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

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

// MarketManager manages market data and pool information
type MarketManager struct {
	config        *config.UniswapConfig
	logger        *logger.Logger
	pools         map[string]*PoolData
	mu            sync.RWMutex
	cacheGroup    singleflight.Group
	cacheDuration time.Duration
	maxCacheSize  int
}

// PoolData represents data for a Uniswap V3 pool
type PoolData 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
}

// NewMarketManager creates a new market manager
func NewMarketManager(cfg *config.UniswapConfig, logger *logger.Logger) *MarketManager {
	return &MarketManager{
		config:        cfg,
		logger:        logger,
		pools:         make(map[string]*PoolData),
		cacheDuration: time.Duration(cfg.Cache.Expiration) * time.Second,
		maxCacheSize:  cfg.Cache.MaxSize,
	}
}

// GetPool retrieves pool data, either from cache or by fetching it
func (mm *MarketManager) GetPool(ctx context.Context, poolAddress common.Address) (*PoolData, error) {
	// Check if we have it in cache and it's still valid
	poolKey := poolAddress.Hex()

	mm.mu.RLock()
	if pool, exists := mm.pools[poolKey]; exists {
		// Check if cache is still valid
		if time.Since(pool.LastUpdated) < mm.cacheDuration {
			mm.mu.RUnlock()
			return pool, nil
		}
	}
	mm.mu.RUnlock()

	// Use singleflight to prevent duplicate requests for the same pool
	result, err, _ := mm.cacheGroup.Do(poolKey, func() (interface{}, error) {
		return mm.fetchPoolData(ctx, poolAddress)
	})

	if err != nil {
		return nil, err
	}

	pool := result.(*PoolData)

	// Update cache
	mm.mu.Lock()
	// Check if we need to evict old entries
	if len(mm.pools) >= mm.maxCacheSize {
		mm.evictOldest()
	}
	mm.pools[poolKey] = pool
	mm.mu.Unlock()

	return pool, nil
}

// fetchPoolData fetches pool data from the blockchain
func (mm *MarketManager) fetchPoolData(ctx context.Context, poolAddress common.Address) (*PoolData, error) {
	// Connect to Ethereum client
	client, err := ethclient.Dial(mm.config.RPCEndpoint)
	if err != nil {
		return nil, fmt.Errorf("failed to connect to Ethereum node: %v", err)
	}
	defer client.Close()

	// Create Uniswap V3 pool contract instance
	poolContract, err := uniswapv3.NewUniswapV3Pool(poolAddress, client)
	if err != nil {
		return nil, fmt.Errorf("failed to create pool contract instance: %v", err)
	}

	// Fetch pool data concurrently
	var wg sync.WaitGroup
	var token0, token1 common.Address
	var fee uint32
	var liquidity *big.Int
	var sqrtPriceX96 *big.Int
	var tick int32
	var tickSpacing int32

	var token0Err, token1Err, feeErr, liquidityErr, sqrtPriceX96Err, tickErr, tickSpacingErr error

	// Fetch token0
	wg.Add(1)
	go func() {
		defer wg.Done()
		token0, token0Err = poolContract.Token0(nil)
	}()

	// Fetch token1
	wg.Add(1)
	go func() {
		defer wg.Done()
		token1, token1Err = poolContract.Token1(nil)
	}()

	// Fetch fee
	wg.Add(1)
	go func() {
		defer wg.Done()
		fee, feeErr = poolContract.Fee(nil)
	}()

	// Fetch liquidity
	wg.Add(1)
	go func() {
		defer wg.Done()
		liquidity, liquidityErr = poolContract.Liquidity(nil)
	}()

	// Fetch slot0 (sqrtPriceX96 and tick)
	wg.Add(1)
	go func() {
		defer wg.Done()
		slot0, err := poolContract.Slot0(nil)
		if err != nil {
			sqrtPriceX96Err = err
			tickErr = err
			return
		}
		sqrtPriceX96 = slot0.SqrtPriceX96
		tick = slot0.Tick
	}()

	// Fetch tick spacing
	wg.Add(1)
	go func() {
		defer wg.Done()
		tickSpacing, tickSpacingErr = poolContract.TickSpacing(nil)
	}()

	// Wait for all goroutines to complete
	wg.Wait()

	// Check for errors
	if token0Err != nil {
		return nil, fmt.Errorf("failed to fetch token0: %v", token0Err)
	}
	if token1Err != nil {
		return nil, fmt.Errorf("failed to fetch token1: %v", token1Err)
	}
	if feeErr != nil {
		return nil, fmt.Errorf("failed to fetch fee: %v", feeErr)
	}
	if liquidityErr != nil {
		return nil, fmt.Errorf("failed to fetch liquidity: %v", liquidityErr)
	}
	if sqrtPriceX96Err != nil {
		return nil, fmt.Errorf("failed to fetch sqrtPriceX96: %v", sqrtPriceX96Err)
	}
	if tickErr != nil {
		return nil, fmt.Errorf("failed to fetch tick: %v", tickErr)
	}
	if tickSpacingErr != nil {
		return nil, fmt.Errorf("failed to fetch tick spacing: %v", tickSpacingErr)
	}

	// Convert big.Int values to uint256
	liquidityUint256, overflow := uint256.FromBig(liquidity)
	if overflow {
		return nil, fmt.Errorf("liquidity value overflow")
	}

	sqrtPriceX96Uint256, overflow := uint256.FromBig(sqrtPriceX96)
	if overflow {
		return nil, fmt.Errorf("sqrtPriceX96 value overflow")
	}

	pool := &PoolData{
		Address:      poolAddress,
		Token0:       token0,
		Token1:       token1,
		Fee:          int64(fee),
		Liquidity:    liquidityUint256,
		SqrtPriceX96: sqrtPriceX96Uint256,
		Tick:         int(tick),
		TickSpacing:  int(tickSpacing),
		LastUpdated:  time.Now(),
	}

	mm.logger.Debug(fmt.Sprintf("Fetched pool data for %s", poolAddress.Hex()))

	return pool, nil
}

// evictOldest removes the oldest entry from the cache
func (mm *MarketManager) evictOldest() {
	oldestKey := ""
	var oldestTime time.Time

	for key, pool := range mm.pools {
		if oldestKey == "" || pool.LastUpdated.Before(oldestTime) {
			oldestKey = key
			oldestTime = pool.LastUpdated
		}
	}

	if oldestKey != "" {
		delete(mm.pools, oldestKey)
		mm.logger.Debug(fmt.Sprintf("Evicted pool %s from cache", oldestKey))
	}
}

// UpdatePool updates pool data
func (mm *MarketManager) UpdatePool(poolAddress common.Address, liquidity *uint256.Int, sqrtPriceX96 *uint256.Int, tick int) {
	poolKey := poolAddress.Hex()

	mm.mu.Lock()
	defer mm.mu.Unlock()

	if pool, exists := mm.pools[poolKey]; exists {
		pool.Liquidity = liquidity
		pool.SqrtPriceX96 = sqrtPriceX96
		pool.Tick = tick
		pool.LastUpdated = time.Now()
	} else {
		// Create new pool entry
		pool := &PoolData{
			Address:      poolAddress,
			Liquidity:    liquidity,
			SqrtPriceX96: sqrtPriceX96,
			Tick:         tick,
			LastUpdated:  time.Now(),
		}
		mm.pools[poolKey] = pool
	}
}

// GetPoolsByTokens retrieves pools for a pair of tokens
func (mm *MarketManager) GetPoolsByTokens(token0, token1 common.Address) []*PoolData {
	mm.mu.RLock()
	defer mm.mu.RUnlock()

	pools := make([]*PoolData, 0)

	for _, pool := range mm.pools {
		// Check if this pool contains the token pair
		if (pool.Token0 == token0 && pool.Token1 == token1) ||
			(pool.Token0 == token1 && pool.Token1 == token0) {
			pools = append(pools, pool)
		}
	}

	return pools
}

// GetAllPools returns all cached pools
func (mm *MarketManager) GetAllPools() []*PoolData {
	mm.mu.RLock()
	defer mm.mu.RUnlock()

	pools := make([]*PoolData, 0, len(mm.pools))
	for _, pool := range mm.pools {
		pools = append(pools, pool)
	}

	return pools
}

// ClearCache clears all cached pool data
func (mm *MarketManager) ClearCache() {
	mm.mu.Lock()
	defer mm.mu.Unlock()

	mm.pools = make(map[string]*PoolData)
	mm.logger.Info("Cleared pool cache")
}

// GetCacheStats returns cache statistics
func (mm *MarketManager) GetCacheStats() (int, int) {
	mm.mu.RLock()
	defer mm.mu.RUnlock()

	return len(mm.pools), mm.maxCacheSize
}