2023-07-04 11:25:08 -07:00
|
|
|
package consumer
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"errors"
|
2023-07-13 08:26:38 -07:00
|
|
|
"fmt"
|
2023-07-04 11:25:08 -07:00
|
|
|
|
|
|
|
"github.com/wormhole-foundation/wormhole-explorer/common/domain"
|
|
|
|
"github.com/wormhole-foundation/wormhole-explorer/txtracker/chains"
|
|
|
|
"github.com/wormhole-foundation/wormhole-explorer/txtracker/config"
|
|
|
|
sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
)
|
|
|
|
|
2023-07-11 12:31:45 -07:00
|
|
|
var ErrAlreadyProcessed = errors.New("VAA was already processed")
|
|
|
|
|
2023-07-04 11:25:08 -07:00
|
|
|
// ProcessSourceTxParams is a struct that contains the parameters for the ProcessSourceTx method.
|
|
|
|
type ProcessSourceTxParams struct {
|
|
|
|
ChainId sdk.ChainID
|
|
|
|
VaaId string
|
|
|
|
Emitter string
|
|
|
|
Sequence string
|
|
|
|
TxHash string
|
2023-07-11 12:31:45 -07:00
|
|
|
// Overwrite indicates whether to reprocess a VAA that has already been processed.
|
|
|
|
//
|
|
|
|
// In the context of backfilling, sometimes you want to overwrite old data (e.g.: because
|
|
|
|
// the schema changed).
|
|
|
|
// In the context of the service, you usually don't want to overwrite existing data
|
|
|
|
// to avoid processing the same VAA twice, which would result in performance degradation.
|
|
|
|
Overwrite bool
|
2023-07-04 11:25:08 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
func ProcessSourceTx(
|
|
|
|
ctx context.Context,
|
|
|
|
logger *zap.Logger,
|
|
|
|
rpcServiceProviderSettings *config.RpcProviderSettings,
|
|
|
|
repository *Repository,
|
|
|
|
params *ProcessSourceTxParams,
|
|
|
|
) error {
|
|
|
|
|
2023-07-13 08:26:38 -07:00
|
|
|
if !params.Overwrite {
|
|
|
|
// If the message has already been processed, skip it.
|
|
|
|
//
|
|
|
|
// Sometimes the SQS visibility timeout expires and the message is put back into the queue,
|
|
|
|
// even if the RPC nodes have been hit and data has been written to MongoDB.
|
|
|
|
// In those cases, when we fetch the message for the second time,
|
|
|
|
// we don't want to hit the RPC nodes again for performance reasons.
|
|
|
|
processed, err := repository.AlreadyProcessed(ctx, params.VaaId)
|
|
|
|
if err != nil {
|
2023-07-04 11:25:08 -07:00
|
|
|
return err
|
2023-07-13 08:26:38 -07:00
|
|
|
} else if err == nil && processed {
|
|
|
|
return ErrAlreadyProcessed
|
2023-07-04 11:25:08 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-07-13 08:26:38 -07:00
|
|
|
// Get transaction details from the emitter blockchain
|
|
|
|
txDetail, err := chains.FetchTx(ctx, rpcServiceProviderSettings, params.ChainId, params.TxHash)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("failed to process transaction: %w", err)
|
|
|
|
}
|
|
|
|
|
2023-07-04 11:25:08 -07:00
|
|
|
// Store source transaction details in the database
|
|
|
|
p := UpsertDocumentParams{
|
|
|
|
VaaId: params.VaaId,
|
|
|
|
ChainId: params.ChainId,
|
|
|
|
TxDetail: txDetail,
|
2023-07-13 08:26:38 -07:00
|
|
|
TxStatus: domain.SourceTxStatusConfirmed,
|
2023-07-04 11:25:08 -07:00
|
|
|
}
|
|
|
|
return repository.UpsertDocument(ctx, &p)
|
|
|
|
}
|