chain.go

package backfill

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

	"github.com/synapsecns/sanguine/ethergo/util"

	"github.com/ethereum/go-ethereum/common"
	ethTypes "github.com/ethereum/go-ethereum/core/types"
	"github.com/jpillora/backoff"
	indexerconfig "github.com/synapsecns/sanguine/services/explorer/config/indexer"
	"github.com/synapsecns/sanguine/services/explorer/consumer/fetcher"
	"github.com/synapsecns/sanguine/services/explorer/consumer/parser"
	"github.com/synapsecns/sanguine/services/explorer/db"
	"golang.org/x/sync/errgroup"
)

// ChainBackfiller is an explorer backfiller for a chain.
type ChainBackfiller struct {
	// consumerDB is the database that the backfiller will use to store the events.
	consumerDB db.ConsumerDB
	// bridgeParser is the parser to use to parse bridge events.
	bridgeParser *parser.BridgeParser
	// swapParsers is a map from contract address -> parser.
	swapParsers map[common.Address]*parser.SwapParser
	// messageBusParser is the parser to use to parse message bus events.
	messageBusParser *parser.MessageBusParser
	// cctpParser is the parser to use to parse cctp events.
	cctpParser *parser.CCTPParser
	// rfqParser is the parser to use to parse rfq events.
	rfqParser *parser.RFQParser
	// Fetcher is the Fetcher to use to fetch logs.
	Fetcher fetcher.ScribeFetcher
	// chainConfig is the chain config for the chain.
	chainConfig indexerconfig.ChainConfig
}

type contextKey string

const (
	chainKey contextKey = "chainID"
)

// NewChainBackfiller creates a new backfiller for a chain.
func NewChainBackfiller(consumerDB db.ConsumerDB, bridgeParser *parser.BridgeParser, swapParsers map[common.Address]*parser.SwapParser, messageBusParser *parser.MessageBusParser, cctpParser *parser.CCTPParser, rfqParser *parser.RFQParser, fetcher fetcher.ScribeFetcher, chainConfig indexerconfig.ChainConfig) *ChainBackfiller {
	return &ChainBackfiller{
		consumerDB:       consumerDB,
		bridgeParser:     bridgeParser,
		swapParsers:      swapParsers,
		messageBusParser: messageBusParser,
		cctpParser:       cctpParser,
		rfqParser:        rfqParser,
		Fetcher:          fetcher,
		chainConfig:      chainConfig,
	}
}

// Backfill fetches logs from the GraphQL database, parses them, and stores them in the consumer database.
// nolint:cyclop,gocognit
func (c *ChainBackfiller) Backfill(ctx context.Context, livefill bool, refreshRate int) (err error) {
	chainCtx := context.WithValue(ctx, chainKey, fmt.Sprintf("%d", c.chainConfig.ChainID))
	contractsGroup, contractCtx := errgroup.WithContext(chainCtx)

	if !livefill {
		for i := range c.chainConfig.Contracts {
			contract := c.chainConfig.Contracts[i]
			contractsGroup.Go(func() error {
				err := c.backfillContractLogs(contractCtx, contract)
				if err != nil {
					return fmt.Errorf("could not backfill contract logs: %w", err)
				}
				return nil
			})
		}
	} else {
		for i := range c.chainConfig.Contracts {
			contract := c.chainConfig.Contracts[i]
			contractsGroup.Go(func() error {
				b := &backoff.Backoff{
					Factor: 2,
					Jitter: true,
					Min:    1 * time.Second,
					Max:    3 * time.Second,
				}
				timeout := time.Duration(0)
				for {
					select {
					case <-chainCtx.Done():
						logger.Errorf("livefill of contract %s on chain %d failed: %v", contract.Address, c.chainConfig.ChainID, chainCtx.Err())

						return fmt.Errorf("livefill of contract %s on chain %d failed: %w", contract.Address, c.chainConfig.ChainID, chainCtx.Err())
					case <-time.After(timeout):
						err := c.backfillContractLogs(contractCtx, contract)
						if err != nil {
							timeout = b.Duration()
							logger.Warnf("could not livefill contract %s on chain %d, retrying %v", contract.Address, c.chainConfig.ChainID, err)

							continue
						}
						b.Reset()
						timeout = time.Duration(refreshRate) * time.Second
						logger.Infof("processed range for contract %s on chain %d, continuing to livefill in %d seconds - refresh rate %d ", contract.Address, c.chainConfig.ChainID, timeout, refreshRate)
					}
				}
			})
		}
	}
	if err := contractsGroup.Wait(); err != nil {
		logger.Errorf("error backfilling chain %d completed %v", c.chainConfig.ChainID, err)

		return fmt.Errorf("error while backfilling chain %d: %w", c.chainConfig.ChainID, err)
	}
	return nil
}

// makeEventParser returns a parser for a contract using it's config.
// in the event one is not present, this function will return an error.
func (c *ChainBackfiller) makeEventParser(contract indexerconfig.ContractConfig) (eventParser parser.Parser, err error) {
	contractType, err := indexerconfig.ContractTypeFromString(contract.ContractType)
	if err != nil {
		return nil, fmt.Errorf("could not create event parser for unknown contract type: %s", contract.ContractType)
	}
	switch contractType {
	case indexerconfig.BridgeContractType:
		eventParser = c.bridgeParser
	case indexerconfig.SwapContractType:
		eventParser = c.swapParsers[common.HexToAddress(contract.Address)]
	case indexerconfig.MessageBusContractType:
		eventParser = c.messageBusParser
	case indexerconfig.MetaSwapContractType:
		eventParser = c.swapParsers[common.HexToAddress(contract.Address)]
	case indexerconfig.CCTPContractType:
		eventParser = c.cctpParser
	case indexerconfig.RFQContractType:
		eventParser = c.rfqParser
	}
	return eventParser, nil
}

// backfillContractLogs creates a backfiller for a given contract with an independent context
// nolint:cyclop,gocognit
func (c *ChainBackfiller) backfillContractLogs(parentCtx context.Context, contract indexerconfig.ContractConfig) (err error) {
	// make the event parser
	eventParser, err := c.makeEventParser(contract)
	if err != nil {
		return err
	}

	startHeight := uint64(contract.StartBlock)

	// Set start block to -1 to trigger backfill from last block stored by explorer,
	// otherwise backfilling will begin at the block number specified in the config file.
	if contract.StartBlock < 0 {
		startHeight, err = c.consumerDB.GetLastStoredBlock(parentCtx, c.chainConfig.ChainID, contract.Address)
		if err != nil {
			return fmt.Errorf("could not get last block number: %w, %s", err, contract.ContractType)
		}
	}
	var endHeight uint64
	err = c.retryWithBackoff(parentCtx, func(ctx context.Context) error {
		// TODO change to get last unconfirmed block
		endHeight, err = c.Fetcher.FetchLastIndexed(parentCtx, c.chainConfig.ChainID, contract.Address)
		if err != nil {
			return fmt.Errorf("could not get last indexed height, %w", err)
		}
		return nil
	})
	if err != nil {
		return fmt.Errorf("could not get last indexed for contract %s: %w, %v", contract.Address, err, c.chainConfig)
	}

	currentHeight := startHeight
	// Iterate over all blocks and fetch logs with the current contract address.
	for currentHeight <= endHeight {
		// Create context for backfilling chunks
		g, groupCtx := errgroup.WithContext(parentCtx)

		chunkStart := currentHeight
		chunkEnd := currentHeight + (c.chainConfig.FetchBlockIncrement-1)*uint64(c.chainConfig.MaxGoroutines)
		if chunkEnd > endHeight {
			chunkEnd = endHeight
		}

		iterator := util.NewChunkIterator(big.NewInt(int64(chunkStart)), big.NewInt(int64(chunkEnd)), int(c.chainConfig.FetchBlockIncrement)-1, true)
		for subChunk := iterator.NextChunk(); subChunk != nil; subChunk = iterator.NextChunk() {
			chunkVar := subChunk
			g.Go(func() error {
				b := &backoff.Backoff{
					Factor: 2,
					Jitter: true,
					Min:    1 * time.Second,
					Max:    3 * time.Second,
				}

				timeout := time.Duration(0)

				for {
					select {
					case <-groupCtx.Done():
						return fmt.Errorf("context canceled: %w", groupCtx.Err())
					case <-time.After(timeout):
						rangeEnd := chunkVar.EndBlock.Uint64()

						// Fetch the logs from Scribe.
						logs, err := c.Fetcher.FetchLogsInRange(groupCtx, c.chainConfig.ChainID, chunkVar.StartBlock.Uint64(), rangeEnd, common.HexToAddress(contract.Address))
						if err != nil {
							timeout = b.Duration()
							logger.Warnf("could not fetch logs for chain %d: %v. Retrying in %s", c.chainConfig.ChainID, err, timeout)

							continue
						}

						parsedLogs, err := ProcessLogs(groupCtx, logs, c.chainConfig.ChainID, eventParser)
						if err != nil {
							timeout = b.Duration()
							logger.Warnf("could not process logs for chain %d: %s", c.chainConfig.ChainID, err)
							continue
						}

						if len(parsedLogs) > 0 {
							g.Go(func() error {
								return c.storeParsedLogs(groupCtx, parsedLogs)
							})
						}
						return nil
					}
				}
			})
		}

		if err := g.Wait(); err != nil {
			return fmt.Errorf("error while backfilling chain %d: %w", c.chainConfig.ChainID, err)
		}
		logger.Infof("backfilling contract %s chunk completed, %d to %d", contract.Address, chunkStart, chunkEnd)
		// Store the last block in clickhouse
		err = c.retryWithBackoff(parentCtx, func(ctx context.Context) error {
			err = c.consumerDB.StoreLastBlock(parentCtx, c.chainConfig.ChainID, chunkEnd, contract.Address)
			if err != nil {
				return fmt.Errorf("error storing last block, %w", err)
			}
			return nil
		})
		if err != nil {
			logger.Errorf("could not store last block for chain %d: %s %d, %s, %s", c.chainConfig.ChainID, err, chunkEnd, contract.Address, contract.ContractType)
			return fmt.Errorf("could not store last block for chain %d: %w", c.chainConfig.ChainID, err)
		}
		currentHeight = chunkEnd + 1
	}

	return nil
}

// ProcessLogs processes the logs and stores them in the consumer database.
//
//nolint:gocognit,cyclop
func ProcessLogs(ctx context.Context, logs []ethTypes.Log, chainID uint32, eventParser parser.Parser) (parsedLogs []interface{}, _ error) {
	b := &backoff.Backoff{
		Factor: 2,
		Jitter: true,
		Min:    1 * time.Second,
		Max:    10 * time.Second,
	}

	timeout := time.Duration(0)
	logIdx := 0
	for {
		select {
		case <-ctx.Done():
			return parsedLogs, fmt.Errorf("context canceled: %w", ctx.Err())
		case <-time.After(timeout):
			if logIdx >= len(logs) {
				return parsedLogs, nil
			}
			parsedLog, err := eventParser.Parse(ctx, logs[logIdx], chainID)
			if err != nil || parsedLog == nil {
				// TODO: this should really, REALLY use errors.IS and wrap the underlying error
				if strings.Contains(err.Error(), parser.ErrUnknownTopic) {
					logger.Warnf("could not parse log (ErrUnknownTopic) %d, %s %s blocknumber: %d, %s", chainID, logs[logIdx].TxHash, logs[logIdx].Address, logs[logIdx].BlockNumber, err)
				} else { // retry
					logger.Errorf("could not parse log %d, %s blocknumber: %d, %s", chainID, logs[logIdx].Address, logs[logIdx].BlockNumber, err)
					timeout = b.Duration()
					continue
				}
			}

			parsedLogs = append(parsedLogs, parsedLog)

			logIdx++

			// Reset the backoff after successful log parse run to prevent bloated back offs.
			b.Reset()
			timeout = time.Duration(0)
		}
	}
}

func (c *ChainBackfiller) storeParsedLogs(ctx context.Context, parsedEvents []interface{}) error {
	b := &backoff.Backoff{
		Factor: 2,
		Jitter: true,
		Min:    3 * time.Millisecond,
		Max:    2 * time.Second,
	}
	timeout := time.Duration(0)

	for {
		select {
		case <-ctx.Done():
			return fmt.Errorf("context canceled while storing events: %w", ctx.Err())
		case <-time.After(timeout):
			err := c.consumerDB.StoreEvents(ctx, parsedEvents)
			if err != nil {
				logger.Errorf("Error storing events: %v", err)
				timeout = b.Duration()
				continue
			}
			return nil
		}
	}
}

const maxAttempt = 20

type retryableFunc func(ctx context.Context) error

// retryWithBackoff will retry to get data with a backoff.
func (c *ChainBackfiller) retryWithBackoff(ctx context.Context, doFunc retryableFunc) error {
	b := &backoff.Backoff{
		Factor: 2,
		Jitter: true,
		Min:    1 * time.Second,
		Max:    3 * time.Second,
	}

	timeout := time.Duration(0)
	attempts := 0
	for attempts < maxAttempt {
		select {
		case <-ctx.Done():
			return fmt.Errorf("%w while retrying", ctx.Err())
		case <-time.After(timeout):
			err := doFunc(ctx)
			if err != nil {
				timeout = b.Duration()
				attempts++
			} else {
				return nil
			}
		}
	}
	return fmt.Errorf("max attempts reached while retrying")
}