circle.go

package relayer

import (
	"context"
	"errors"
	"fmt"
	"math/big"

	"github.com/ethereum/go-ethereum/accounts/abi/bind"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/crypto"
	"github.com/synapsecns/sanguine/core/metrics"
	"github.com/synapsecns/sanguine/ethergo/client"
	signerConfig "github.com/synapsecns/sanguine/ethergo/signer/config"
	"github.com/synapsecns/sanguine/ethergo/submitter"
	"github.com/synapsecns/sanguine/services/cctp-relayer/config"
	messagetransmitter "github.com/synapsecns/sanguine/services/cctp-relayer/contracts/messagetransmitter"
	tokenmessenger "github.com/synapsecns/sanguine/services/cctp-relayer/contracts/tokenmessenger"
	db2 "github.com/synapsecns/sanguine/services/cctp-relayer/db"
	relayTypes "github.com/synapsecns/sanguine/services/cctp-relayer/types"
	omniClient "github.com/synapsecns/sanguine/services/omnirpc/client"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"gorm.io/gorm"
)

type circleCCTPHandler struct {
	cfg                      config.Config
	db                       db2.CCTPRelayerDB
	omniRPCClient            omniClient.RPCClient
	boundMessageTransmitters map[uint32]*messagetransmitter.MessageTransmitter
	txSubmitter              submitter.TransactionSubmitter
	relayerAddress           common.Address
	handler                  metrics.Handler
}

// NewCircleCCTPHandler creates a new CircleCCTPHandler.
func NewCircleCCTPHandler(ctx context.Context, cfg config.Config, db db2.CCTPRelayerDB, omniRPCClient omniClient.RPCClient, txSubmitter submitter.TransactionSubmitter, handler metrics.Handler) (CCTPHandler, error) {
	boundMessageTransmitters := make(map[uint32]*messagetransmitter.MessageTransmitter)
	for _, chain := range cfg.Chains {
		cl, err := omniRPCClient.GetConfirmationsClient(ctx, int(chain.ChainID), 1)
		if err != nil {
			return nil, fmt.Errorf("could not get client: %w", err)
		}
		transmitterAddr, err := GetMessageTransmitterAddress(ctx, chain.GetCCTPAddress(), cl)
		if err != nil {
			return nil, fmt.Errorf("could not get message transmitter address: %w", err)
		}
		boundMessageTransmitters[chain.ChainID], err = messagetransmitter.NewMessageTransmitter(transmitterAddr, cl)
		if err != nil {
			return nil, fmt.Errorf("could not build bound contract: %w", err)
		}
	}
	signer, err := signerConfig.SignerFromConfig(ctx, cfg.Signer)
	if err != nil {
		return nil, fmt.Errorf("could not make cctp signer: %w", err)
	}

	return &circleCCTPHandler{
		cfg:                      cfg,
		db:                       db,
		omniRPCClient:            omniRPCClient,
		boundMessageTransmitters: boundMessageTransmitters,
		txSubmitter:              txSubmitter,
		relayerAddress:           signer.Address(),
		handler:                  handler,
	}, nil
}

func (c *circleCCTPHandler) HandleLog(ctx context.Context, log *types.Log, chainID uint32) (processQueue bool, err error) {
	if log == nil {
		return false, fmt.Errorf("log is nil")
	}

	// shouldn't be possible: maybe remove?
	if len(log.Topics) == 0 {
		return false, fmt.Errorf("not enough topics")
	}

	switch log.Topics[0] {
	case tokenmessenger.DepositForBurnTopic:
		msg, err := c.handleDepositForBurn(ctx, log, chainID)
		if err != nil {
			return false, fmt.Errorf("could not store depositForBurn: %w", err)
		}
		if msg != nil {
			processQueue = true
		}
		return processQueue, nil
	case messagetransmitter.MessageReceivedTopic:
		err = c.handleMessageReceived(ctx, log, chainID)
		if err != nil {
			return false, fmt.Errorf("could not handle message received: %w", err)
		}
		return false, nil
	default:
		logger.Warnf("unknown topic %s", log.Topics[0])
		return false, nil
	}
}

func (c *circleCCTPHandler) FetchAndProcessSentEvent(parentCtx context.Context, txHash common.Hash, chainID uint32) (msg *relayTypes.Message, err error) {
	ctx, span := c.handler.Tracer().Start(parentCtx, "FetchAndProcessSentEvent", trace.WithAttributes(
		attribute.String(metrics.TxHash, txHash.String()),
		attribute.Int(metrics.ChainID, int(chainID)),
	))

	defer func() {
		metrics.EndSpanWithErr(span, err)
	}()

	// check if message already exists before we do anything
	msg, err = c.db.GetMessageByOriginHash(ctx, txHash)
	// if we already have the message, we can just return it
	if err == nil {
		return msg, nil
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, fmt.Errorf("could not get message by origin hash: %w", err)
	}

	ethClient, err := c.omniRPCClient.GetConfirmationsClient(ctx, int(chainID), 1)
	if err != nil {
		return nil, fmt.Errorf("could not get chain client: %w", err)
	}

	// TODO: consider pulling from scribe?
	// TODO: this function is particularly prone to silent errors
	receipt, err := ethClient.TransactionReceipt(ctx, txHash)
	if err != nil {
		return nil, fmt.Errorf("could not get transaction receipt: %w", err)
	}

	for _, log := range receipt.Logs {
		// this should never happen
		if len(log.Topics) == 0 {
			continue
		}

		if log.Topics[0] == messagetransmitter.MessageSentTopic {
			msg, err = c.handleDepositForBurn(ctx, log, chainID)
			if err != nil {
				return nil, fmt.Errorf("could not handle depositForBurn: %w", err)
			}
			return msg, nil
		}
	}
	return nil, fmt.Errorf("no depositForBurn event found")
}

func (c *circleCCTPHandler) SubmitReceiveMessage(parentCtx context.Context, msg *relayTypes.Message) (err error) {
	ctx, span := c.handler.Tracer().Start(parentCtx, "SubmitReceiveMessage", trace.WithAttributes(
		attribute.String("message_hash", msg.MessageHash),
		attribute.Int(metrics.ChainID, int(msg.DestChainID)),
	))

	defer func() {
		metrics.EndSpanWithErr(span, err)
	}()

	contract, ok := c.boundMessageTransmitters[msg.DestChainID]
	if !ok {
		return fmt.Errorf("no contract found for chain %d", msg.DestChainID)
	}

	//TODO: check if already executed on destination?

	var nonce uint64
	var destTxHash common.Hash
	nonce, err = c.txSubmitter.SubmitTransaction(ctx, big.NewInt(int64(msg.DestChainID)), func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) {
		tx, err = contract.ReceiveMessage(transactor, msg.Message, msg.Attestation)
		if err != nil {
			return nil, fmt.Errorf("could not submit transaction: %w", err)
		}

		destTxHash = tx.Hash()
		return tx, nil
	})
	if err != nil {
		err = fmt.Errorf("could not submit transaction: %w", err)
		return err
	}
	span.SetAttributes(
		attribute.String("dest_tx_hash", destTxHash.String()),
		attribute.Int("nonce", int(nonce)),
	)

	// Store the completed message.
	// Note: this can cause double submission sometimes
	msg.State = relayTypes.Submitted
	msg.DestNonce = int(nonce)
	msg.DestTxHash = destTxHash.String()
	err = c.db.StoreMessage(ctx, *msg)
	if err != nil {
		return fmt.Errorf("could not store completed message: %w", err)
	}

	return nil
}

func (c *circleCCTPHandler) handleDepositForBurn(parentCtx context.Context, log *types.Log, chainID uint32) (msg *relayTypes.Message, err error) {
	ctx, span := c.handler.Tracer().Start(parentCtx, "handleDepositForBurn", trace.WithAttributes(
		attribute.String(metrics.TxHash, log.TxHash.Hex()),
		attribute.Int(metrics.ChainID, int(chainID)),
		attribute.Int("block_number", int(log.BlockNumber)),
	))

	defer func() {
		metrics.EndSpanWithErr(span, err)
	}()

	ethClient, err := c.omniRPCClient.GetConfirmationsClient(ctx, int(chainID), 1)
	if err != nil {
		return nil, fmt.Errorf("could not get chain client: %w", err)
	}

	eventParser, err := tokenmessenger.NewTokenMessengerFilterer(log.Address, ethClient)
	if err != nil {
		return nil, fmt.Errorf("could not create event parser: %w", err)
	}

	event, err := eventParser.ParseDepositForBurn(*log)
	if err != nil {
		return nil, fmt.Errorf("could not parse depositForBurn: %w", err)
	}

	// check that we sent the tx
	if event.Depositor != c.relayerAddress {
		span.AddEvent(fmt.Sprintf("depositor %s does not match relayer address %s", event.Depositor.String(), c.relayerAddress.String()))
		//nolint:nilnil
		return nil, nil
	}
	span.SetAttributes(
		attribute.Int("dest_domain", int(event.DestinationDomain)),
		attribute.String("depositor", event.Depositor.Hex()),
	)

	message, err := c.getMessagePayload(ctx, log.TxHash, chainID, ethClient)
	if err != nil {
		return nil, fmt.Errorf("could not get message payload: %w", err)
	}

	destChainID, err := CircleDomainToChainID(event.DestinationDomain, IsTestnetChainID(chainID))
	if err != nil {
		return nil, fmt.Errorf("could not convert circle domain to chain ID: %w", err)
	}

	sourceDomain, err := c.boundMessageTransmitters[chainID].LocalDomain(&bind.CallOpts{Context: ctx})
	if err != nil {
		return nil, fmt.Errorf("could not get local domain: %w", err)
	}

	rawMsg := relayTypes.Message{
		RequestID:     GetCircleRequestID(sourceDomain, event.Nonce),
		OriginTxHash:  log.TxHash.Hex(),
		OriginChainID: chainID,
		DestChainID:   destChainID,
		Message:       message,
		MessageHash:   crypto.Keccak256Hash(message).String(),
		BlockNumber:   log.BlockNumber,
	}

	span.SetAttributes(
		attribute.String("request_id", rawMsg.RequestID),
		attribute.String("message_hash", rawMsg.MessageHash),
		attribute.Int(metrics.Origin, int(rawMsg.OriginChainID)),
		attribute.Int(metrics.Destination, int(rawMsg.DestChainID)),
	)

	// Store the requested message.
	rawMsg.State = relayTypes.Pending
	err = c.db.StoreMessage(ctx, rawMsg)
	if err != nil {
		return nil, fmt.Errorf("could not store pending message: %w", err)
	}

	return &rawMsg, nil
}

//nolint:cyclop
func (c *circleCCTPHandler) handleMessageReceived(parentCtx context.Context, log *types.Log, chainID uint32) (err error) {
	ctx, span := c.handler.Tracer().Start(parentCtx, "handleMessageReceived", trace.WithAttributes(
		attribute.String(metrics.TxHash, log.TxHash.Hex()),
		attribute.Int(metrics.ChainID, int(chainID)),
		attribute.Int("block_number", int(log.BlockNumber)),
	))
	defer func() {
		metrics.EndSpanWithErr(span, err)
	}()

	ethClient, err := c.omniRPCClient.GetConfirmationsClient(ctx, int(chainID), 1)
	if err != nil {
		return fmt.Errorf("could not get chain client: %w", err)
	}
	eventParser, err := messagetransmitter.NewMessageTransmitterFilterer(log.Address, ethClient)
	if err != nil {
		return fmt.Errorf("could not create event parser: %w", err)
	}
	event, err := eventParser.ParseMessageReceived(*log)
	if err != nil {
		return fmt.Errorf("could not parse circle request fulfilled: %w", err)
	}

	requestID := GetCircleRequestID(event.SourceDomain, event.Nonce)
	span.SetAttributes(
		attribute.Int("source_domain", int(event.SourceDomain)),
		attribute.String("request_id", requestID),
	)

	msg, err := c.db.GetMessageByRequestID(ctx, requestID)
	if err != nil {
		span.AddEvent(fmt.Sprintf("could not get message with id %s; dropping", requestID))
		return nil
	}
	if msg == nil {
		return fmt.Errorf("no message found")
	}

	// Mark as Complete and store the message.
	msg.State = relayTypes.Complete
	msg.DestTxHash = log.TxHash.String()
	err = c.db.StoreMessage(ctx, *msg)
	if err != nil {
		return fmt.Errorf("could not store complete message: %w", err)
	}
	span.AddEvent("message marked as complete")
	return nil
}

// getMessagePayload gets the message payload from a MessageTransmitter event by fetching the transaction receipt.
func (c *circleCCTPHandler) getMessagePayload(ctx context.Context, txHash common.Hash, chainID uint32, ethClient client.EVM) (message []byte, err error) {
	chainCfg, err := c.cfg.GetChainConfig(chainID)
	if err != nil {
		return nil, fmt.Errorf("could not get chain config: %w", err)
	}
	transmitterAddr, err := GetMessageTransmitterAddress(ctx, chainCfg.GetCCTPAddress(), ethClient)
	if err != nil {
		return nil, fmt.Errorf("could not get message transmitter address: %w", err)
	}
	parser, err := messagetransmitter.NewMessageTransmitterFilterer(transmitterAddr, ethClient)
	if err != nil {
		return nil, fmt.Errorf("could not create depositForBurn event parser: %w", err)
	}
	receipt, err := ethClient.TransactionReceipt(ctx, txHash)
	if err != nil {
		return nil, fmt.Errorf("could not get transaction receipt: %w", err)
	}
	for _, log := range receipt.Logs {
		if log.Topics[0] == messagetransmitter.MessageSentTopic {
			messageSentEvent, err := parser.ParseMessageSent(*log)
			if err != nil {
				return nil, fmt.Errorf("could not parse depositForBurn event: %w", err)
			}
			return messageSentEvent.Message, nil
		}
	}
	return nil, fmt.Errorf("no depositForBurn event found")
}