wormhole-explorer/parser/processor/processor.go

package processor

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

	"github.com/wormhole-foundation/wormhole-explorer/common/client/alert"
	vaaPayloadParser "github.com/wormhole-foundation/wormhole-explorer/common/client/parser"
	"github.com/wormhole-foundation/wormhole-explorer/common/domain"
	parserAlert "github.com/wormhole-foundation/wormhole-explorer/parser/internal/alert"
	"github.com/wormhole-foundation/wormhole-explorer/parser/internal/metrics"
	"github.com/wormhole-foundation/wormhole-explorer/parser/parser"
	sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
	"go.uber.org/zap"
)

type Processor struct {
	parser        vaaPayloadParser.ParserVAAAPIClient
	repository    *parser.Repository
	alert         alert.AlertClient
	metrics       metrics.Metrics
	tokenProvider *domain.TokenProvider
	logger        *zap.Logger
}

func New(parser vaaPayloadParser.ParserVAAAPIClient, repository *parser.Repository, alert alert.AlertClient, metrics metrics.Metrics, tokenProvider *domain.TokenProvider, logger *zap.Logger) *Processor {
	return &Processor{
		parser:        parser,
		repository:    repository,
		alert:         alert,
		metrics:       metrics,
		tokenProvider: tokenProvider,
		logger:        logger,
	}
}

func (p *Processor) Process(ctx context.Context, params *Params) (*parser.ParsedVaaUpdate, error) {
	// unmarshal vaa.
	vaa, err := sdk.Unmarshal(params.Vaa)
	if err != nil {
		return nil, err
	}

	// call vaa-payload-parser api to parse a VAA.
	chainID := uint16(vaa.EmitterChain)
	emitterAddress := vaa.EmitterAddress.String()
	sequence := fmt.Sprintf("%d", vaa.Sequence)

	p.metrics.IncVaaPayloadParserRequestCount(chainID)
	vaaParseResponse, err := p.parser.ParseVaaWithStandarizedProperties(vaa)
	if err != nil {
		// split metrics error not found and others errors.
		if errors.Is(err, vaaPayloadParser.ErrNotFound) {
			p.metrics.IncVaaPayloadParserNotFoundCount(chainID)
		} else {
			p.metrics.IncVaaPayloadParserErrorCount(chainID)
		}

		// if error is ErrInternalError or ErrCallEndpoint return error in order to retry.
		if errors.Is(err, vaaPayloadParser.ErrInternalError) || errors.Is(err, vaaPayloadParser.ErrCallEndpoint) {
			// send alert when exists and error calling vaa-payload-parser component.
			alertContext := alert.AlertContext{
				Details: map[string]string{
					"trackID":        params.TrackID,
					"chainID":        vaa.EmitterChain.String(),
					"emitterAddress": emitterAddress,
					"sequence":       sequence,
				},
				Error: err,
			}
			p.alert.CreateAndSend(ctx, parserAlert.AlertKeyVaaPayloadParserError, alertContext)
			return nil, err
		}

		p.logger.Info("VAA cannot be parsed", zap.Error(err),
			zap.String("trackId", params.TrackID),
			zap.Uint16("chainId", chainID),
			zap.String("address", emitterAddress),
			zap.String("sequence", sequence))
		return nil, nil
	}
	p.metrics.IncVaaPayloadParserSuccessCount(chainID)
	p.metrics.IncVaaParsed(chainID)

	standardizedProperties := p.transformStandarizedProperties(params.TrackID, vaa.MessageID(), vaaParseResponse.StandardizedProperties)

	// create ParsedVaaUpdate to upsert.
	now := time.Now()
	vaaParsed := parser.ParsedVaaUpdate{
		ID:                        vaa.MessageID(),
		EmitterChain:              vaa.EmitterChain,
		EmitterAddr:               emitterAddress,
		Sequence:                  sequence,
		AppIDs:                    standardizedProperties.AppIds,
		ParsedPayload:             vaaParseResponse.ParsedPayload,
		RawStandardizedProperties: vaaParseResponse.StandardizedProperties,
		StandardizedProperties:    standardizedProperties,
		Timestamp:                 vaa.Timestamp,
		UpdatedAt:                 &now,
	}

	err = p.repository.UpsertParsedVaa(ctx, vaaParsed)
	if err != nil {
		p.logger.Error("Error inserting vaa in repository",
			zap.String("trackId", params.TrackID),
			zap.String("id", vaaParsed.ID),
			zap.Error(err))
		// send alert when exists and error inserting parsed vaa.
		alertContext := alert.AlertContext{
			Details: map[string]string{
				"trackID":        params.TrackID,
				"chainID":        vaa.EmitterChain.String(),
				"emitterAddress": emitterAddress,
				"sequence":       sequence,
				"appIDs":         strings.Join(standardizedProperties.AppIds, ", "),
			},
			Error: err}
		p.alert.CreateAndSend(ctx, parserAlert.AlertKeyInsertParsedVaaError, alertContext)
		return nil, err
	}
	p.metrics.IncVaaParsedInserted(chainID)

	p.logger.Info("parsed VAA was successfully persisted", zap.String("trackId", params.TrackID), zap.String("id", vaaParsed.ID))
	return &vaaParsed, nil
}

// transformStandarizedProperties transform amount and fee amount.
func (p *Processor) transformStandarizedProperties(trackID, vaaID string, sp vaaPayloadParser.StandardizedProperties) vaaPayloadParser.StandardizedProperties {
	// transform amount.
	amount := p.transformAmount(sp.TokenChain, trackID, sp.TokenAddress, sp.Amount, vaaID)
	// transform fee amount.
	feeAmount := p.transformAmount(sp.FeeChain, trackID, sp.FeeAddress, sp.Fee, vaaID)
	// create StandardizedProperties.
	return vaaPayloadParser.StandardizedProperties{
		AppIds:       sp.AppIds,
		FromChain:    sp.FromChain,
		FromAddress:  sp.FromAddress,
		ToChain:      sp.ToChain,
		ToAddress:    sp.ToAddress,
		TokenChain:   sp.TokenChain,
		TokenAddress: sp.TokenAddress,
		Amount:       amount,
		FeeAddress:   sp.FeeAddress,
		FeeChain:     sp.FeeChain,
		Fee:          feeAmount,
	}
}

// transformAmount transform amount and fee amount.
func (p *Processor) transformAmount(chainID sdk.ChainID, trackID, nativeAddress, amount, vaaID string) string {

	if chainID == sdk.ChainIDUnset || nativeAddress == "" || amount == "" {
		return ""
	}

	nativeHex, err := domain.DecodeNativeAddressToHex(sdk.ChainID(chainID), nativeAddress)
	if err != nil {
		p.logger.Warn("Native address cannot be transformed to hex",
			zap.String("trackId", trackID),
			zap.String("vaaId", vaaID),
			zap.String("nativeAddress", nativeAddress),
			zap.Uint16("chain", uint16(chainID)))
		return ""
	}

	addr, err := sdk.StringToAddress(nativeHex)
	if err != nil {
		p.logger.Warn("Address cannot be parsed",
			zap.String("trackId", trackID),
			zap.String("vaaId", vaaID),
			zap.String("nativeAddress", nativeAddress),
			zap.Uint16("chain", uint16(chainID)))
		return ""
	}
	// Get the token metadata
	//
	// This is complementary data about the token that is not present in the VAA itself.
	tokenMeta, ok := p.tokenProvider.GetTokenByAddress(sdk.ChainID(chainID), addr.String())
	if !ok {
		p.logger.Warn("Token metadata not found",
			zap.String("trackId", trackID),
			zap.String("vaaId", vaaID),
			zap.String("nativeAddress", nativeAddress),
			zap.Uint16("chain", uint16(chainID)))
		return ""
	}

	bigAmount := new(big.Int)
	bigAmount, ok = bigAmount.SetString(amount, 10)
	if !ok {
		p.logger.Error("Cannot parse amount",
			zap.String("trackId", trackID),
			zap.String("vaaId", vaaID),
			zap.String("amount", amount),
			zap.String("nativeAddress", nativeAddress),
			zap.Uint16("chain", uint16(chainID)))
		return ""
	}

	if tokenMeta.Decimals < 8 {
		// factor = 10 ^ (8 - tokenMeta.Decimals)
		var factor big.Int
		factor.Exp(big.NewInt(10), big.NewInt(int64(8-tokenMeta.Decimals)), nil)

		bigAmount = bigAmount.Mul(bigAmount, &factor)
	}

	return bigAmount.String()
}

// createStandarizedProperties create a new StandardizedProperties with amount and fee amount transformed.
func createStandarizedProperties(m vaaPayloadParser.StandardizedProperties, amount, feeAmount, fromAddress, toAddress, tokenAddress, feeAddress string) vaaPayloadParser.StandardizedProperties {
	return vaaPayloadParser.StandardizedProperties{
		AppIds:       m.AppIds,
		FromChain:    m.FromChain,
		FromAddress:  fromAddress,
		ToChain:      m.ToChain,
		ToAddress:    toAddress,
		TokenChain:   m.TokenChain,
		TokenAddress: tokenAddress,
		Amount:       amount,
		FeeAddress:   feeAddress,
		FeeChain:     m.FeeChain,
		Fee:          feeAmount,
	}
}