Merge b28a201ff1 into 9d280f7d5f

common/client/guardian/guardian.go

@@ -0,0 +1,65 @@
+package guardian
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+const DefaultTimeout = 10
+
+// GuardianAPIClient guardian api client.
+type GuardianAPIClient struct {
+	Client  http.Client
+	BaseURL string
+	Logger  *zap.Logger
+}
+
+// NewGuardianAPIClient create new instances of GuardianAPIClient.
+func NewGuardianAPIClient(timeout int64, baseURL string, logger *zap.Logger) (GuardianAPIClient, error) {
+	if timeout == 0 {
+		timeout = DefaultTimeout
+	}
+	if baseURL == "" {
+		return GuardianAPIClient{}, errors.New("baseURL can not be empty")
+	}
+
+	return GuardianAPIClient{
+		Client: http.Client{
+			Timeout: time.Duration(timeout) * time.Second,
+		},
+		BaseURL: baseURL,
+		Logger:  logger,
+	}, nil
+}
+
+// SignedVaa represent a guardianAPI signed vaa response.
+type SignedVaa struct {
+	VaaBytes []byte `json:"vaaBytes"`
+}
+
+// GetSignedVAA get signed vaa.
+func (c *GuardianAPIClient) GetSignedVAA(vaaID string) (*SignedVaa, error) {
+	endpointUrl := fmt.Sprintf("%s/v1/signed_vaa/%s", c.BaseURL, vaaID)
+	resp, err := c.Client.Get(endpointUrl)
+	if err != nil {
+		c.Logger.Error("failed to call endpoint", zap.String("endpoint", endpointUrl), zap.Error(err))
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		c.Logger.Error("failed to call endpoint", zap.String("endpoint", endpointUrl), zap.Int("status_code", resp.StatusCode))
+		return nil, errors.New("failed to call endpoint, status code is not 200")
+	}
+
+	var signedVaa SignedVaa
+	err = json.NewDecoder(resp.Body).Decode(&signedVaa)
+	if err != nil {
+		c.Logger.Error("failed to decode response", zap.Error(err))
+		return nil, err
+	}
+	return &signedVaa, nil
+}

fly-event-processor/.gitignore

@@ -0,0 +1,2 @@
+.idea/
+.vscode/

fly-event-processor/DockerFile

@@ -0,0 +1,20 @@
+# syntax=docker.io/docker/dockerfile:1.3@sha256:42399d4635eddd7a9b8a24be879d2f9a930d0ed040a61324cfdf59ef1357b3b2
+FROM --platform=linux/amd64 docker.io/golang:1.20.7-bullseye@sha256:74b09b3b6fa5aa542df8ef974cb745eb477be72f6fcf821517fb410aff532b00 AS build
+WORKDIR /app
+
+COPY fly-event-processor fly-event-processor
+COPY common common
+
+# Build the Go app
+RUN cd fly-event-processor && CGO_ENABLED=0 GOOS=linux go build -o "./fly-event-processor" cmd/main.go
+
+############################
+# STEP 2 build a small image
+############################
+FROM alpine
+#Copy certificates
+COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
+# Copy our static executable.
+COPY --from=build "/app/fly-event-processor/fly-event-processor" "/fly-event-processor"
+# Run the binary.
+ENTRYPOINT ["/fly-event-processor"]

fly-event-processor/Makefile

@@ -0,0 +1,11 @@
+SHELL := /bin/bash
+
+
+build:
+	go build -o bin/service cmd/main.go
+	
+test:
+	go test -v -cover ./...
+
+
+.PHONY: build doc test

fly-event-processor/Readme.md

@@ -0,0 +1 @@
+# FlyEventProcessor

fly-event-processor/cmd/main.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/cmd/service"
+)
+
+func main() {
+	execute()
+}
+
+func execute() error {
+	root := &cobra.Command{
+		Use: "fly-event-processor",
+		Run: func(cmd *cobra.Command, args []string) {
+			if len(args) == 0 {
+				service.Run()
+			}
+		},
+	}
+
+	addServiceCommand(root)
+
+	return root.Execute()
+}
+
+func addServiceCommand(root *cobra.Command) {
+	serviceCommand := &cobra.Command{
+		Use:   "service",
+		Short: "Run fly-event-processor as service",
+		Run: func(_ *cobra.Command, _ []string) {
+			service.Run()
+		},
+	}
+	root.AddCommand(serviceCommand)
+}

fly-event-processor/cmd/service/run.go

@@ -0,0 +1,215 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/wormhole-foundation/wormhole-explorer/common/client/sqs"
+	"github.com/wormhole-foundation/wormhole-explorer/common/dbutil"
+	"github.com/wormhole-foundation/wormhole-explorer/common/health"
+	"github.com/wormhole-foundation/wormhole-explorer/common/logger"
+	"github.com/wormhole-foundation/wormhole-explorer/common/pool"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/http/vaa"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/processor"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/queue"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/storage"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.uber.org/zap"
+
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/config"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/consumer"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/http/infrastructure"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/internal/metrics"
+)
+
+func Run() {
+	rootCtx, rootCtxCancel := context.WithCancel(context.Background())
+
+	// load config
+	cfg, err := config.New(rootCtx)
+	if err != nil {
+		log.Fatal("Error loading config: ", err)
+	}
+
+	// initialize metrics
+	metrics := newMetrics(cfg)
+
+	// build logger
+	logger := logger.New("wormholescan-fly-event-processor", logger.WithLevel(cfg.LogLevel))
+	logger.Info("Starting wormholescan-fly-event-processor ...")
+
+	// create guardian provider pool
+	guardianApiProviderPool, err := newGuardianProviderPool(cfg)
+	if err != nil {
+		logger.Fatal("Error creating guardian provider pool: ", zap.Error(err))
+	}
+
+	// initialize the database client
+	db, err := dbutil.Connect(rootCtx, logger, cfg.MongoURI, cfg.MongoDatabase, false)
+	if err != nil {
+		log.Fatal("Failed to initialize MongoDB client: ", err)
+	}
+
+	// create a new repository
+	repository := storage.NewRepository(logger, db.Database)
+
+	// create a new processor
+	processor := processor.NewProcessor(guardianApiProviderPool, repository, logger, metrics)
+
+	// start serving /health and /ready endpoints
+	healthChecks, err := makeHealthChecks(rootCtx, cfg, db.Database)
+	if err != nil {
+		logger.Fatal("Failed to create health checks", zap.Error(err))
+	}
+	vaaCtrl := vaa.NewController(processor.Process, repository, logger)
+	server := infrastructure.NewServer(logger, cfg.Port, vaaCtrl, cfg.PprofEnabled, healthChecks...)
+	server.Start()
+
+	// create and start a duplicate VAA consumer.
+	duplicateVaaConsumeFunc := newDuplicateVaaConsumeFunc(rootCtx, cfg, metrics, logger)
+	duplicateVaa := consumer.New(duplicateVaaConsumeFunc, processor.Process, logger, metrics, cfg.P2pNetwork, cfg.ConsumerWorkerSize)
+	duplicateVaa.Start(rootCtx)
+
+	logger.Info("Started wormholescan-fly-event-processor")
+
+	// Waiting for signal
+	sigterm := make(chan os.Signal, 1)
+	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
+	select {
+	case <-rootCtx.Done():
+		logger.Warn("Terminating with root context cancelled.")
+	case signal := <-sigterm:
+		logger.Info("Terminating with signal.", zap.String("signal", signal.String()))
+	}
+
+	// graceful shutdown
+	logger.Info("Cancelling root context...")
+	rootCtxCancel()
+
+	logger.Info("Closing Http server...")
+	server.Stop()
+
+	logger.Info("Closing MongoDB connection...")
+	db.DisconnectWithTimeout(10 * time.Second)
+
+	logger.Info("Terminated wormholescan-fly-event-processor")
+
+}
+
+func newAwsConfig(ctx context.Context, cfg *config.ServiceConfiguration) (aws.Config, error) {
+
+	region := cfg.AwsRegion
+
+	if cfg.AwsAccessKeyID != "" && cfg.AwsSecretAccessKey != "" {
+
+		credentials := credentials.NewStaticCredentialsProvider(cfg.AwsAccessKeyID, cfg.AwsSecretAccessKey, "")
+
+		customResolver := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) {
+			if cfg.AwsEndpoint != "" {
+				return aws.Endpoint{
+					PartitionID:   "aws",
+					URL:           cfg.AwsEndpoint,
+					SigningRegion: region,
+				}, nil
+			}
+
+			return aws.Endpoint{}, &aws.EndpointNotFoundError{}
+		})
+
+		awsCfg, err := awsconfig.LoadDefaultConfig(
+			ctx,
+			awsconfig.WithRegion(region),
+			awsconfig.WithEndpointResolver(customResolver),
+			awsconfig.WithCredentialsProvider(credentials),
+		)
+		return awsCfg, err
+	}
+	return awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
+}
+
+func newSqsConsumer(ctx context.Context, cfg *config.ServiceConfiguration, sqsUrl string) (*sqs.Consumer, error) {
+
+	awsconfig, err := newAwsConfig(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	consumer, err := sqs.NewConsumer(
+		awsconfig,
+		sqsUrl,
+		sqs.WithMaxMessages(10),
+		sqs.WithVisibilityTimeout(60),
+	)
+	return consumer, err
+}
+
+func makeHealthChecks(
+	ctx context.Context,
+	cfg *config.ServiceConfiguration,
+	db *mongo.Database,
+) ([]health.Check, error) {
+
+	awsConfig, err := newAwsConfig(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	plugins := []health.Check{
+		health.SQS(awsConfig, cfg.DuplicateVaaSQSUrl),
+		health.Mongo(db),
+	}
+
+	return plugins, nil
+}
+
+func newMetrics(cfg *config.ServiceConfiguration) metrics.Metrics {
+	if !cfg.MetricsEnabled {
+		return metrics.NewDummyMetrics()
+	}
+	return metrics.NewPrometheusMetrics(cfg.Environment)
+}
+
+func newGuardianProviderPool(cfg *config.ServiceConfiguration) (*pool.Pool, error) {
+	if cfg.GuardianAPIConfigurationJson == nil {
+		return nil, errors.New("guardian api provider configuration is missing")
+	}
+
+	var guardianCfgs []pool.Config
+	for _, provider := range cfg.GuardianAPIConfigurationJson.GuardianProviders {
+		guardianCfgs = append(guardianCfgs, pool.Config{
+			Id:                provider.ProviderUrl,
+			Description:       provider.ProviderName,
+			RequestsPerMinute: provider.RequestsPerMinute,
+			Priority:          provider.Priority,
+		})
+	}
+
+	if len(guardianCfgs) == 0 {
+		return nil, errors.New("guardian api provider configuration is empty")
+	}
+	return pool.NewPool(guardianCfgs), nil
+}
+
+func newDuplicateVaaConsumeFunc(
+	ctx context.Context,
+	cfg *config.ServiceConfiguration,
+	metrics metrics.Metrics,
+	logger *zap.Logger,
+) queue.ConsumeFunc {
+
+	sqsConsumer, err := newSqsConsumer(ctx, cfg, cfg.DuplicateVaaSQSUrl)
+	if err != nil {
+		logger.Fatal("failed to create sqs consumer", zap.Error(err))
+	}
+
+	vaaQueue := queue.NewEventSqs(sqsConsumer, metrics, logger)
+	return vaaQueue.Consume
+}

fly-event-processor/config/config.go

@@ -0,0 +1,83 @@
+package config
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/joho/godotenv"
+	"github.com/sethvargo/go-envconfig"
+)
+
+// p2p network constants.
+const (
+	P2pMainNet = "mainnet"
+	P2pTestNet = "testnet"
+	P2pDevNet  = "devnet"
+)
+
+// ServiceConfiguration represents the application configuration when running as service with default values.
+type ServiceConfiguration struct {
+	// Global configuration
+	Environment    string `env:"ENVIRONMENT,required"`
+	LogLevel       string `env:"LOG_LEVEL,default=INFO"`
+	Port           string `env:"PORT,default=8000"`
+	PprofEnabled   bool   `env:"PPROF_ENABLED,default=false"`
+	P2pNetwork     string `env:"P2P_NETWORK,required"`
+	AlertEnabled   bool   `env:"ALERT_ENABLED,default=false"`
+	AlertApiKey    string `env:"ALERT_API_KEY"`
+	MetricsEnabled bool   `env:"METRICS_ENABLED,default=false"`
+	// Fly event consumer configuration
+	ConsumerWorkerSize int `env:"CONSUMER_WORKER_SIZE,default=1"`
+	// Database configuration
+	MongoURI      string `env:"MONGODB_URI,required"`
+	MongoDatabase string `env:"MONGODB_DATABASE,required"`
+	// AWS configuration
+	AwsEndpoint        string `env:"AWS_ENDPOINT"`
+	AwsAccessKeyID     string `env:"AWS_ACCESS_KEY_ID"`
+	AwsSecretAccessKey string `env:"AWS_SECRET_ACCESS_KEY"`
+	AwsRegion          string `env:"AWS_REGION"`
+	DuplicateVaaSQSUrl string `env:"DUPLICATE_VAA_SQS_URL"`
+	// Guardian api provider configuration
+	GuardianAPIProviderPath       string `env:"GUARDIAN_API_PROVIDER_PATH,required"`
+	*GuardianAPIConfigurationJson `required:"false"`
+}
+
+type GuardianAPIConfigurationJson struct {
+	GuardianProviders []GuardianProvider `json:"guardian_providers"`
+}
+
+type GuardianProvider struct {
+	ProviderName      string `json:"name"`
+	ProviderUrl       string `json:"url"`
+	RequestsPerMinute uint16 `json:"requests_per_minute"`
+	Priority          uint8  `json:"priority"`
+}
+
+// New creates a configuration with the values from .env file and environment variables.
+func New(ctx context.Context) (*ServiceConfiguration, error) {
+	_ = godotenv.Load(".env", "../.env")
+
+	var configuration ServiceConfiguration
+	if err := envconfig.Process(ctx, &configuration); err != nil {
+		return nil, err
+	}
+
+	// Load guardian api provider configuration.
+	if configuration.GuardianAPIProviderPath != "" {
+		guardianAPIJsonFile, err := os.ReadFile(configuration.GuardianAPIProviderPath)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read guardian API provider settings from file: %w", err)
+		}
+		var GuardianAPIConfigurationJson GuardianAPIConfigurationJson
+		if err := json.Unmarshal(guardianAPIJsonFile, &GuardianAPIConfigurationJson); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal guardian API provider settings: %w", err)
+		}
+		configuration.GuardianAPIConfigurationJson = &GuardianAPIConfigurationJson
+	} else {
+		return nil, fmt.Errorf("guardian API provider settings file is required")
+	}
+
+	return &configuration, nil
+}

fly-event-processor/consumer/consumer.go

@@ -0,0 +1,99 @@
+package consumer
+
+import (
+	"context"
+
+	"github.com/wormhole-foundation/wormhole-explorer/common/pool"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/internal/metrics"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/processor"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/queue"
+	sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
+	"go.uber.org/zap"
+)
+
+// Consumer consumer struct definition.
+type Consumer struct {
+	consumeFunc  queue.ConsumeFunc
+	processor    processor.ProcessorFunc
+	guardianPool *pool.Pool
+	logger       *zap.Logger
+	metrics      metrics.Metrics
+	p2pNetwork   string
+	workersSize  int
+}
+
+// New creates a new vaa consumer.
+func New(
+	consumeFunc queue.ConsumeFunc,
+	processor processor.ProcessorFunc,
+	logger *zap.Logger,
+	metrics metrics.Metrics,
+	p2pNetwork string,
+	workersSize int,
+) *Consumer {
+
+	c := Consumer{
+		consumeFunc: consumeFunc,
+		processor:   processor,
+		logger:      logger,
+		metrics:     metrics,
+		p2pNetwork:  p2pNetwork,
+		workersSize: workersSize,
+	}
+
+	return &c
+}
+
+// Start consumes messages from VAA queue, parse and store those messages in a repository.
+func (c *Consumer) Start(ctx context.Context) {
+	ch := c.consumeFunc(ctx)
+	for i := 0; i < c.workersSize; i++ {
+		go c.producerLoop(ctx, ch)
+	}
+}
+
+func (c *Consumer) producerLoop(ctx context.Context, ch <-chan queue.ConsumerMessage) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case msg := <-ch:
+			c.processEvent(ctx, msg)
+		}
+	}
+}
+
+func (c *Consumer) processEvent(ctx context.Context, msg queue.ConsumerMessage) {
+	event := msg.Data()
+	vaaID := event.Data.VaaID
+	chainID := sdk.ChainID(event.Data.ChainID)
+
+	logger := c.logger.With(
+		zap.String("trackId", event.TrackID),
+		zap.String("vaaId", vaaID))
+
+	if msg.IsExpired() {
+		msg.Failed()
+		logger.Debug("event is expired")
+		c.metrics.IncDuplicatedVaaExpired(chainID)
+		return
+	}
+
+	params := &processor.Params{
+		TrackID: event.TrackID,
+		VaaID:   vaaID,
+		ChainID: chainID,
+	}
+
+	err := c.processor(ctx, params)
+	if err != nil {
+		msg.Failed()
+		logger.Error("error processing event", zap.Error(err))
+		c.metrics.IncDuplicatedVaaFailed(chainID)
+		return
+	}
+
+	msg.Done()
+	logger.Debug("event processed")
+	c.metrics.IncDuplicatedVaaProcessed(chainID)
+}

fly-event-processor/go.mod

@@ -0,0 +1,81 @@
+module github.com/wormhole-foundation/wormhole-explorer/fly-event-processor
+
+go 1.20
+
+require (
+	github.com/ansrivas/fiberprometheus/v2 v2.6.1
+	github.com/aws/aws-sdk-go-v2 v1.17.5
+	github.com/aws/aws-sdk-go-v2/config v1.18.15
+	github.com/aws/aws-sdk-go-v2/credentials v1.13.15
+	github.com/gofiber/fiber/v2 v2.52.4
+	github.com/joho/godotenv v1.5.1
+	github.com/sethvargo/go-envconfig v1.0.0
+	github.com/spf13/cobra v1.8.0
+	github.com/wormhole-foundation/wormhole-explorer/common v0.0.0-20240422172607-688a0d0f718e
+	go.mongodb.org/mongo-driver v1.11.2
+	go.uber.org/zap v1.27.0
+)
+
+require (
+	github.com/andybalholm/brotli v1.0.5 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.23 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.29 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.23 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.30 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.23 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sns v1.20.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sqs v1.20.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.12.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.18.5 // indirect
+	github.com/aws/smithy-go v1.13.5 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/btcsuite/btcd/btcec/v2 v2.3.2 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
+	github.com/deepmap/oapi-codegen v1.8.2 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/ethereum/go-ethereum v1.10.21 // indirect
+	github.com/fsnotify/fsnotify v1.6.0 // indirect
+	github.com/go-redis/redis/v8 v8.11.5 // indirect
+	github.com/gofiber/adaptor/v2 v2.2.1 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/uuid v1.5.0 // indirect
+	github.com/holiman/uint256 v1.2.1 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/influxdata/influxdb-client-go/v2 v2.12.2 // indirect
+	github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097 // indirect
+	github.com/klauspost/compress v1.17.2 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mattn/go-runewidth v0.0.15 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect
+	github.com/onsi/gomega v1.30.0 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/prometheus/client_golang v1.16.0 // indirect
+	github.com/prometheus/client_model v0.4.0 // indirect
+	github.com/prometheus/common v0.44.0 // indirect
+	github.com/prometheus/procfs v0.11.1 // indirect
+	github.com/rivo/uniseg v0.4.4 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/valyala/bytebufferpool v1.0.0 // indirect
+	github.com/valyala/fasthttp v1.51.0 // indirect
+	github.com/valyala/tcplisten v1.0.0 // indirect
+	github.com/wormhole-foundation/wormhole/sdk v0.0.0-20240416174455-25e60611a867 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
+	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/crypto v0.19.0 // indirect
+	golang.org/x/net v0.21.0 // indirect
+	golang.org/x/sync v0.4.0 // indirect
+	golang.org/x/sys v0.17.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/time v0.3.0 // indirect
+	google.golang.org/protobuf v1.32.0 // indirect
+)
+
+replace github.com/wormhole-foundation/wormhole-explorer/common => ../common

fly-event-processor/go.sum

@@ -0,0 +1,269 @@
+github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs=
+github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
+github.com/ansrivas/fiberprometheus/v2 v2.6.1 h1:wac3pXaE6BYYTF04AC6K0ktk6vCD+MnDOJZ3SK66kXM=
+github.com/ansrivas/fiberprometheus/v2 v2.6.1/go.mod h1:MloIKvy4yN6hVqlRpJ/jDiR244YnWJaQC0FIqS8A+MY=
+github.com/aws/aws-sdk-go-v2 v1.17.4/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw=
+github.com/aws/aws-sdk-go-v2 v1.17.5 h1:TzCUW1Nq4H8Xscph5M/skINUitxM5UBAyvm2s7XBzL4=
+github.com/aws/aws-sdk-go-v2 v1.17.5/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw=
+github.com/aws/aws-sdk-go-v2/config v1.18.15 h1:509yMO0pJUGUugBP2H9FOFyV+7Mz7sRR+snfDN5W4NY=
+github.com/aws/aws-sdk-go-v2/config v1.18.15/go.mod h1:vS0tddZqpE8cD9CyW0/kITHF5Bq2QasW9Y1DFHD//O0=
+github.com/aws/aws-sdk-go-v2/credentials v1.13.15 h1:0rZQIi6deJFjOEgHI9HI2eZcLPPEGQPictX66oRFLL8=
+github.com/aws/aws-sdk-go-v2/credentials v1.13.15/go.mod h1:vRMLMD3/rXU+o6j2MW5YefrGMBmdTvkLLGqFwMLBHQc=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.23 h1:Kbiv9PGnQfG/imNI4L/heyUXvzKmcWSBeDvkrQz5pFc=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.23/go.mod h1:mOtmAg65GT1HIL/HT/PynwPbS+UG0BgCZ6vhkPqnxWo=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.28/go.mod h1:3lwChorpIM/BhImY/hy+Z6jekmN92cXGPI1QJasVPYY=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.29 h1:9/aKwwus0TQxppPXFmf010DFrE+ssSbzroLVYINA+xE=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.29/go.mod h1:Dip3sIGv485+xerzVv24emnjX5Sg88utCL8fwGmCeWg=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.22/go.mod h1:EqK7gVrIGAHyZItrD1D8B0ilgwMD1GiWAmbU4u/JHNk=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.23 h1:b/Vn141DBuLVgXbhRWIrl9g+ww7G+ScV5SzniWR13jQ=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.23/go.mod h1:mr6c4cHC+S/MMkrjtSlG4QA36kOznDep+0fga5L/fGQ=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.3.30 h1:IVx9L7YFhpPq0tTnGo8u8TpluFu7nAn9X3sUDMb11c0=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.3.30/go.mod h1:vsbq62AOBwQ1LJ/GWKFxX8beUEYeRp/Agitrxee2/qM=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.23 h1:QoOybhwRfciWUBbZ0gp9S7XaDnCuSTeK/fySB99V1ls=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.23/go.mod h1:9uPh+Hrz2Vn6oMnQYiUi/zbh3ovbnQk19YKINkQny44=
+github.com/aws/aws-sdk-go-v2/service/sns v1.20.2 h1:MU/v2qtfGjKexJ09BMqE8pXo9xYMhT13FXjKgFc0cFw=
+github.com/aws/aws-sdk-go-v2/service/sns v1.20.2/go.mod h1:VN2n9SOMS1lNbh5YD7o+ho0/rgfifSrK//YYNiVVF5E=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.20.2 h1:CSNIo1jiw7KrkdgZjCOnotu6yuB3IybhKLuSQrTLNfo=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.20.2/go.mod h1:1ttxGjUHZliCQMpPss1sU5+Ph/5NvdMFRzr96bv8gm0=
+github.com/aws/aws-sdk-go-v2/service/sso v1.12.4 h1:qJdM48OOLl1FBSzI7ZrA1ZfLwOyCYqkXV5lko1hYDBw=
+github.com/aws/aws-sdk-go-v2/service/sso v1.12.4/go.mod h1:jtLIhd+V+lft6ktxpItycqHqiVXrPIRjWIsFIlzMriw=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.4 h1:YRkWXQveFb0tFC0TLktmmhGsOcCgLwvq88MC2al47AA=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.4/go.mod h1:zVwRrfdSmbRZWkUkWjOItY7SOalnFnq/Yg2LVPqDjwc=
+github.com/aws/aws-sdk-go-v2/service/sts v1.18.5 h1:L1600eLr0YvTT7gNh3Ni24yGI7NSHkq9Gp62vijPRCs=
+github.com/aws/aws-sdk-go-v2/service/sts v1.18.5/go.mod h1:1mKZHLLpDMHTNSYPJ7qrcnCQdHCWsNQaT0xRvq2u80s=
+github.com/aws/smithy-go v1.13.5 h1:hgz0X/DX0dGqTYpGALqXJoRKRj5oQ7150i5FdTePzO8=
+github.com/aws/smithy-go v1.13.5/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA=
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/btcsuite/btcd/btcec/v2 v2.3.2 h1:5n0X6hX0Zk+6omWcihdYvdAlGf2DfasC0GMf7DClJ3U=
+github.com/btcsuite/btcd/btcec/v2 v2.3.2/go.mod h1:zYzJ8etWJQIv1Ogk7OzpWjowwOdXY1W/17j2MW85J04=
+github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 h1:q0rUy8C/TYNBQS1+CGKw68tLOFYSNEs0TFnxxnS9+4U=
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cpuguy83/go-md2man/v2 v2.0.3/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/cyberdelia/templates v0.0.0-20141128023046-ca7fffd4298c/go.mod h1:GyV+0YP4qX0UQ7r2MoYZ+AvYDp12OF5yg4q8rGnyNh4=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/decred/dcrd/crypto/blake256 v1.0.1 h1:7PltbUIQB7u/FfZ39+DGa/ShuMyJ5ilcvdfma9wOH6Y=
+github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 h1:8UrgZ3GkP4i/CLijOJx79Yu+etlyjdBU4sfcs2WYQMs=
+github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0/go.mod h1:v57UDF4pDQJcEfFUCRop3lJL149eHGSe9Jvczhzjo/0=
+github.com/deepmap/oapi-codegen v1.8.2 h1:SegyeYGcdi0jLLrpbCMoJxnUUn8GBXHsvr4rbzjuhfU=
+github.com/deepmap/oapi-codegen v1.8.2/go.mod h1:YLgSKSDv/bZQB7N4ws6luhozi3cEdRktEqrX88CvjIw=
+github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/ethereum/go-ethereum v1.10.21 h1:5lqsEx92ZaZzRyOqBEXux4/UR06m296RGzN3ol3teJY=
+github.com/ethereum/go-ethereum v1.10.21/go.mod h1:EYFyF19u3ezGLD4RqOkLq+ZCXzYbLoNDdZlMt7kyKFg=
+github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
+github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
+github.com/getkin/kin-openapi v0.61.0/go.mod h1:7Yn5whZr5kJi6t+kShccXS8ae1APpYTW6yheSwk8Yi4=
+github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
+github.com/go-chi/chi/v5 v5.0.0/go.mod h1:BBug9lr0cqtdAhsu6R4AAdvufI0/XBzAQSsUqJpoZOs=
+github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
+github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
+github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
+github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+github.com/gofiber/adaptor/v2 v2.2.1 h1:givE7iViQWlsTR4Jh7tB4iXzrlKBgiraB/yTdHs9Lv4=
+github.com/gofiber/adaptor/v2 v2.2.1/go.mod h1:AhR16dEqs25W2FY/l8gSj1b51Azg5dtPDmm+pruNOrc=
+github.com/gofiber/fiber/v2 v2.52.4 h1:P+T+4iK7VaqUsq2PALYEfBBo6bJZ4q3FP8cZ84EggTM=
+github.com/gofiber/fiber/v2 v2.52.4/go.mod h1:KEOE+cXMhXG0zHc9d8+E38hoX+ZN7bhOtgeF2oT6jrQ=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
+github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golangci/lint-1 v0.0.0-20181222135242-d2cdd8c08219/go.mod h1:/X8TswGSh1pIozq4ZwCfxS0WA5JGXguxk94ar/4c87Y=
+github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
+github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
+github.com/holiman/uint256 v1.2.1 h1:XRtyuda/zw2l+Bq/38n5XUoEF72aSOu/77Thd9pPp2o=
+github.com/holiman/uint256 v1.2.1/go.mod h1:y4ga/t+u+Xwd7CpDgZESaRcWy0I7XMlTMA25ApIH5Jw=
+github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
+github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/influxdata/influxdb-client-go/v2 v2.12.2 h1:uYABKdrEKlYm+++qfKdbgaHKBPmoWR5wpbmj6MBB/2g=
+github.com/influxdata/influxdb-client-go/v2 v2.12.2/go.mod h1:YteV91FiQxRdccyJ2cHvj2f/5sq4y4Njqu1fQzsQCOU=
+github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097 h1:vilfsDSy7TDxedi9gyBkMvAirat/oRcL0lFdJBf6tdM=
+github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
+github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
+github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
+github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
+github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
+github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
+github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/labstack/echo/v4 v4.2.1/go.mod h1:AA49e0DZ8kk5jTOOCKNuPR6oTnBS0dYiM4FW1e6jwpg=
+github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
+github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
+github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
+github.com/matryer/moq v0.0.0-20190312154309-6cfb0558e1bd/go.mod h1:9ELz6aaclSIGnZBoaSLZ3NAl1VTufbOrXBPvtcy6WiQ=
+github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-colorable v0.1.7/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
+github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
+github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
+github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
+github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
+github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
+github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
+github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZgg3U=
+github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
+github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
+github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe h1:iruDEfMl2E6fbMZ9s0scYfZQ84/6SPL6zC8ACM2oIL0=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
+github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
+github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
+github.com/onsi/gomega v1.30.0 h1:hvMK7xYz4D3HapigLTeGdId/NcfQx1VHMJc60ew99+8=
+github.com/onsi/gomega v1.30.0/go.mod h1:9sxs+SwGrKI0+PWe4Fxa9tFQQBG5xSsSbMXOI8PPpoQ=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
+github.com/prometheus/client_golang v1.16.0 h1:yk/hx9hDbrGHovbci4BY+pRMfSuuat626eFsHb7tmT8=
+github.com/prometheus/client_golang v1.16.0/go.mod h1:Zsulrv/L9oM40tJ7T815tM89lFEugiJ9HzIqaAx4LKc=
+github.com/prometheus/client_model v0.4.0 h1:5lQXD3cAg1OXBf4Wq03gTrXHeaV0TQvGfUooCfx1yqY=
+github.com/prometheus/client_model v0.4.0/go.mod h1:oMQmHW1/JoDwqLtg57MGgP/Fb1CJEYF2imWWhWtMkYU=
+github.com/prometheus/common v0.44.0 h1:+5BrQJwiBB9xsMygAB3TNvpQKOwlkc25LbISbrdOOfY=
+github.com/prometheus/common v0.44.0/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
+github.com/prometheus/procfs v0.11.1 h1:xRC8Iq1yyca5ypa9n1EZnWZkt7dwcoRPQwX/5gwaUuI=
+github.com/prometheus/procfs v0.11.1/go.mod h1:eesXgaPo1q7lBpVMoMy0ZOFTth9hBn4W/y0/p/ScXhY=
+github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
+github.com/rivo/uniseg v0.4.4 h1:8TfxU8dW6PdqD27gjM8MVNuicgxIjxpm4K7x4jp8sis=
+github.com/rivo/uniseg v0.4.4/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/sethvargo/go-envconfig v1.0.0 h1:1C66wzy4QrROf5ew4KdVw942CQDa55qmlYmw9FZxZdU=
+github.com/sethvargo/go-envconfig v1.0.0/go.mod h1:Lzc75ghUn5ucmcRGIdGQ33DKJrcjk4kihFYgSTBmjIc=
+github.com/spf13/cobra v1.8.0 h1:7aJaZx1B85qltLMc546zn58BxxfZdR/W22ej9CFoEf0=
+github.com/spf13/cobra v1.8.0/go.mod h1:WXLWApfZ71AjXPya3WOlMsY9yMs7YeiHhFVlvLyhcho=
+github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
+github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
+github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
+github.com/valyala/fasthttp v1.51.0 h1:8b30A5JlZ6C7AS81RsWjYMQmrZG6feChmgAolCl1SqA=
+github.com/valyala/fasthttp v1.51.0/go.mod h1:oI2XroL+lI7vdXyYoQk03bXBThfFl2cVdIA3Xl7cH8g=
+github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
+github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
+github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVSA8=
+github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
+github.com/wormhole-foundation/wormhole/sdk v0.0.0-20240416174455-25e60611a867 h1:GXUBP09C/bnEukdU6H2AY81d0m8UWrWEejDp6CgiFQA=
+github.com/wormhole-foundation/wormhole/sdk v0.0.0-20240416174455-25e60611a867/go.mod h1:pE/jYet19kY4P3V6mE2+01zvEfxdyBqv6L6HsnSa5uc=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+go.mongodb.org/mongo-driver v1.11.2 h1:+1v2rDQUWNcGW7/7E0Jvdz51V38XXxJfhzbV17aNHCw=
+go.mongodb.org/mongo-driver v1.11.2/go.mod h1:s7p5vEtfbeR1gYi6pnj3c3/urpbLv2T5Sfd6Rp2HBB8=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.19.0 h1:ENy+Az/9Y1vSrlrvBSyna3PITt4tiZLf7sgCjZBX7Wo=
+golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
+golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
+golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
+golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.4.0 h1:zxkM55ReGkDlKSM+Fu41A+zmbZuaPVbGMzvvdUPznYQ=
+golang.org/x/sync v0.4.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200826173525-f9321e4c35a6/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
+golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/time v0.0.0-20201208040808-7e3f01d25324/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
+golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191125144606-a911d9008d1f/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
+google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

fly-event-processor/guardian/guardian.go

@@ -0,0 +1,20 @@
+package guardian
+
+import (
+	client "github.com/wormhole-foundation/wormhole-explorer/common/client/guardian"
+	"github.com/wormhole-foundation/wormhole-explorer/common/pool"
+)
+
+// GuardianAPIClient is a wrapper around the Guardian API client and the pool of providers.
+type GuardianAPIClient struct {
+	Client *client.GuardianAPIClient
+	Pool   *pool.Pool
+}
+
+// NewGuardianAPIClient creates a new Guardian API client.
+func NewGuardianAPIClient(client *client.GuardianAPIClient, pool *pool.Pool) *GuardianAPIClient {
+	return &GuardianAPIClient{
+		Client: client,
+		Pool:   pool,
+	}
+}

fly-event-processor/http/infrastructure/server.go

@@ -0,0 +1,53 @@
+package infrastructure
+
+import (
+	"github.com/ansrivas/fiberprometheus/v2"
+	"github.com/gofiber/fiber/v2"
+	"github.com/gofiber/fiber/v2/middleware/pprof"
+	health "github.com/wormhole-foundation/wormhole-explorer/common/health"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/http/vaa"
+	"go.uber.org/zap"
+)
+
+type Server struct {
+	app    *fiber.App
+	port   string
+	logger *zap.Logger
+}
+
+func NewServer(logger *zap.Logger, port string, vaaController *vaa.Controller, pprofEnabled bool, checks ...health.Check) *Server {
+	app := fiber.New(fiber.Config{DisableStartupMessage: true})
+	prometheus := fiberprometheus.New("wormscan-fly-event-processor")
+	prometheus.RegisterAt(app, "/metrics")
+
+	// config use of middlware.
+	if pprofEnabled {
+		app.Use(pprof.New())
+	}
+	app.Use(prometheus.Middleware)
+
+	ctrl := health.NewController(checks, logger)
+	api := app.Group("/api")
+	api.Get("/health", ctrl.HealthCheck)
+	api.Get("/ready", ctrl.ReadyCheck)
+	api.Post("/vaa/duplicated", vaaController.Process)
+	return &Server{
+		app:    app,
+		port:   port,
+		logger: logger,
+	}
+}
+
+// Start listen serves HTTP requests from addr.
+func (s *Server) Start() {
+	addr := ":" + s.port
+	s.logger.Info("Listening on " + addr)
+	go func() {
+		s.app.Listen(addr)
+	}()
+}
+
+// Stop gracefull server.
+func (s *Server) Stop() {
+	_ = s.app.Shutdown()
+}

fly-event-processor/http/vaa/controller.go

@@ -0,0 +1,55 @@
+package vaa
+
+import (
+	"fmt"
+
+	"github.com/gofiber/fiber/v2"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/processor"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/storage"
+	"go.uber.org/zap"
+)
+
+// Controller definition.
+type Controller struct {
+	logger     *zap.Logger
+	repository *storage.Repository
+	processor  processor.ProcessorFunc
+}
+
+// NewController creates a Controller instance.
+// func NewController(repository *Repository, processor processor.ProcessorFunc, logger *zap.Logger) *Controller {
+func NewController(processor processor.ProcessorFunc, repository *storage.Repository, logger *zap.Logger) *Controller {
+	return &Controller{processor: processor, repository: repository, logger: logger}
+}
+
+// Process processes the VAA message.
+func (c *Controller) Process(ctx *fiber.Ctx) error {
+	request := struct {
+		VaaID string `json:"vaaId"`
+	}{}
+
+	if err := ctx.BodyParser(&request); err != nil {
+		c.logger.Error("error parsing request", zap.Error(err))
+		return err
+	}
+
+	vaa, err := c.repository.FindVAAById(ctx.Context(), request.VaaID)
+	if err != nil {
+		c.logger.Error("error getting vaa from collection", zap.Error(err))
+		return err
+	}
+
+	params := processor.Params{
+		TrackID: fmt.Sprintf("controller-%s", request.VaaID),
+		VaaID:   request.VaaID,
+		ChainID: vaa.EmitterChain,
+	}
+
+	err = c.processor(ctx.Context(), &params)
+	if err != nil {
+		c.logger.Error("error processing vaa", zap.Error(err))
+		return err
+	}
+
+	return ctx.JSON(fiber.Map{"message": "success"})
+}

fly-event-processor/internal/metrics/dummy.go

@@ -0,0 +1,26 @@
+package metrics
+
+import sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
+
+// DummyMetrics is a dummy implementation of Metric interface.
+type DummyMetrics struct{}
+
+// NewDummyMetrics returns a new instance of DummyMetrics.
+func NewDummyMetrics() *DummyMetrics {
+	return &DummyMetrics{}
+}
+
+// IncDuplicatedVaaConsumedQueue dummy implementation.
+func (d *DummyMetrics) IncDuplicatedVaaConsumedQueue() {}
+
+// IncDuplicatedVaaProcessed dummy implementation.
+func (d *DummyMetrics) IncDuplicatedVaaProcessed(chainID sdk.ChainID) {}
+
+// IncDuplicatedVaaFailed dummy implementation.
+func (d *DummyMetrics) IncDuplicatedVaaFailed(chainID sdk.ChainID) {}
+
+// IncDuplicatedVaaExpired dummy implementation.
+func (d *DummyMetrics) IncDuplicatedVaaExpired(chainID sdk.ChainID) {}
+
+// IncDuplicatedVaaCanNotFixed dummy implementation.
+func (d *DummyMetrics) IncDuplicatedVaaCanNotFixed(chainID sdk.ChainID) {}

fly-event-processor/internal/metrics/metrics.go

@@ -0,0 +1,13 @@
+package metrics
+
+import sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
+
+const serviceName = "wormscan-fly-event-processor"
+
+type Metrics interface {
+	IncDuplicatedVaaConsumedQueue()
+	IncDuplicatedVaaProcessed(chainID sdk.ChainID)
+	IncDuplicatedVaaFailed(chainID sdk.ChainID)
+	IncDuplicatedVaaExpired(chainID sdk.ChainID)
+	IncDuplicatedVaaCanNotFixed(chainID sdk.ChainID)
+}

fly-event-processor/internal/metrics/prometheus.go

@@ -0,0 +1,51 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
+)
+
+// PrometheusMetrics is a Prometheus implementation of Metric interface.
+type PrometheusMetrics struct {
+	duplicatedVaaCount *prometheus.CounterVec
+}
+
+// NewPrometheusMetrics returns a new instance of PrometheusMetrics.
+func NewPrometheusMetrics(environment string) *PrometheusMetrics {
+	return &PrometheusMetrics{
+		duplicatedVaaCount: promauto.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "wormscan_fly_event_processor_duplicated_vaa_count",
+				Help: "The total number of duplicated VAA processed",
+				ConstLabels: map[string]string{
+					"environment": environment,
+					"service":     serviceName,
+				},
+			}, []string{"chain", "type"}),
+	}
+}
+
+func (m *PrometheusMetrics) IncDuplicatedVaaConsumedQueue() {
+	m.duplicatedVaaCount.WithLabelValues("all", "consumed_queue").Inc()
+}
+
+func (m *PrometheusMetrics) IncDuplicatedVaaProcessed(chainID sdk.ChainID) {
+	chain := chainID.String()
+	m.duplicatedVaaCount.WithLabelValues(chain, "processed").Inc()
+}
+
+func (m *PrometheusMetrics) IncDuplicatedVaaFailed(chainID sdk.ChainID) {
+	chain := chainID.String()
+	m.duplicatedVaaCount.WithLabelValues(chain, "failed").Inc()
+}
+
+func (m *PrometheusMetrics) IncDuplicatedVaaExpired(chainID sdk.ChainID) {
+	chain := chainID.String()
+	m.duplicatedVaaCount.WithLabelValues(chain, "expired").Inc()
+}
+
+func (m *PrometheusMetrics) IncDuplicatedVaaCanNotFixed(chainID sdk.ChainID) {
+	chain := chainID.String()
+	m.duplicatedVaaCount.WithLabelValues(chain, "can_not_fixed").Inc()
+}

fly-event-processor/processor/processor.go

@@ -0,0 +1,220 @@
+package processor
+
+import (
+	"errors"
+	"time"
+
+	"github.com/wormhole-foundation/wormhole-explorer/common/client/guardian"
+	"github.com/wormhole-foundation/wormhole-explorer/common/pool"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/internal/metrics"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/storage"
+	sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
+	"go.uber.org/zap"
+	"golang.org/x/net/context"
+)
+
+type Processor struct {
+	guardianPool *pool.Pool
+	repository   *storage.Repository
+	logger       *zap.Logger
+	metrics      metrics.Metrics
+}
+
+func NewProcessor(
+	guardianPool *pool.Pool,
+	repository *storage.Repository,
+	logger *zap.Logger,
+	metrics metrics.Metrics,
+) *Processor {
+	return &Processor{
+		guardianPool: guardianPool,
+		repository:   repository,
+		logger:       logger,
+		metrics:      metrics,
+	}
+}
+
+func (p *Processor) Process(ctx context.Context, params *Params) error {
+	logger := p.logger.With(
+		zap.String("trackId", params.TrackID),
+		zap.String("vaaId", params.VaaID))
+
+	// 1. check if the vaa stored in the VAA collections is the correct one.
+
+	// 1.1 get vaa from Vaas collection
+	vaaDoc, err := p.repository.FindVAAById(ctx, params.VaaID)
+	if err != nil {
+		logger.Error("error getting vaa from collection", zap.Error(err))
+		return err
+	}
+
+	// 1.2 if the event time has not reached the finality time, the event fail and
+	// will be reprocesed on the next retry.
+	finalityTime := getFinalityTimeByChainID(params.ChainID)
+	if vaaDoc.Timestamp == nil {
+		logger.Error("vaa timestamp is nil")
+		return errors.New("vaa timestamp is nil")
+	}
+
+	vaaTimestamp := *vaaDoc.Timestamp
+	reachedFinalityTime := time.Now().After(vaaTimestamp.Add(finalityTime))
+	if !reachedFinalityTime {
+		logger.Debug("event time has not reached the finality time",
+			zap.Time("finalityTime", vaaTimestamp.Add(finalityTime)))
+		return errors.New("event time has not reached the finality time")
+	}
+
+	// 1.3 call guardian api to get signed_vaa.
+	guardians := p.guardianPool.GetItems()
+	var signedVaa *guardian.SignedVaa
+	for _, g := range guardians {
+		g.Wait(ctx)
+		guardianAPIClient, err := guardian.NewGuardianAPIClient(
+			guardian.DefaultTimeout,
+			g.Id,
+			logger)
+		if err != nil {
+			logger.Error("error creating guardian api client", zap.Error(err))
+			continue
+		}
+		signedVaa, err = guardianAPIClient.GetSignedVAA(params.VaaID)
+		if err != nil {
+			logger.Error("error getting signed vaa from guardian api", zap.Error(err))
+			continue
+		}
+		break
+	}
+
+	if signedVaa == nil {
+		logger.Error("error getting signed vaa from guardian api")
+		return errors.New("error getting signed vaa from guardian api")
+	}
+
+	// 1.4 compare digest from vaa and signedVaa
+	guardianVAA, err := sdk.Unmarshal(signedVaa.VaaBytes)
+	if err != nil {
+		logger.Error("error unmarshalling guardian signed vaa", zap.Error(err))
+		return err
+	}
+
+	vaa, err := sdk.Unmarshal(vaaDoc.Vaa)
+	if err != nil {
+		logger.Error("error unmarshalling vaa", zap.Error(err))
+		return err
+	}
+
+	// If the guardian digest is the same that the vaa digest,
+	// the stored vaa in the vaas collection is the correct one.
+	if guardianVAA.HexDigest() == vaa.HexDigest() {
+		logger.Debug("vaa stored in Vaas collections is the correct")
+		return nil
+	}
+
+	// 2. Check for each duplicate VAAs to detect which is the correct one.
+
+	// 2.1 This check is necessary to avoid race conditions when the vaa is processed
+	if vaaDoc.TxHash == "" {
+		logger.Error("vaa txHash is empty")
+		return errors.New("vaa txHash is empty")
+	}
+
+	// 2.2 Get all duplicate vaas by vaaId
+	duplicateVaaDocs, err := p.repository.FindDuplicateVAAs(ctx, params.TrackID)
+	if err != nil {
+		logger.Error("error getting duplicate vaas from collection", zap.Error(err))
+		return err
+	}
+
+	// 2.3 Check each duplicate VAA to detect which is the correct one.
+	for _, duplicateVaaDoc := range duplicateVaaDocs {
+		duplicateVaa, err := sdk.Unmarshal(duplicateVaaDoc.Vaa)
+		if err != nil {
+			logger.Error("error unmarshalling vaa", zap.Error(err))
+			return err
+		}
+
+		if guardianVAA.HexDigest() == duplicateVaa.HexDigest() {
+			err := p.repository.FixVAA(ctx, params.VaaID, duplicateVaaDoc.ID)
+			if err != nil {
+				logger.Error("error fixing vaa", zap.Error(err))
+				return err
+			}
+			logger.Debug("vaa fixed")
+			return nil
+		}
+	}
+
+	logger.Debug("can't fix duplicate vaa")
+	p.metrics.IncDuplicatedVaaCanNotFixed(params.ChainID)
+	return errors.New("can't fix duplicate vaa")
+}
+
+func getFinalityTimeByChainID(chainID sdk.ChainID) time.Duration {
+	// Time to finalize for each chain.
+	// ref: https://docs.wormhole.com/wormhole/reference/constants
+	switch chainID {
+	case sdk.ChainIDSolana:
+		return 14 * time.Second
+	case sdk.ChainIDEthereum:
+		return 975 * time.Second
+	case sdk.ChainIDTerra:
+		return 6 * time.Second
+	case sdk.ChainIDBSC:
+		return 48 * time.Second
+	case sdk.ChainIDPolygon:
+		return 66 * time.Second
+	case sdk.ChainIDAvalanche:
+		return 2 * time.Second
+	case sdk.ChainIDOasis:
+		return 12 * time.Second
+	case sdk.ChainIDAlgorand:
+		return 4 * time.Second
+	case sdk.ChainIDFantom:
+		return 5 * time.Second
+	case sdk.ChainIDKarura:
+		return 24 * time.Second
+	case sdk.ChainIDAcala:
+		return 24 * time.Second
+	case sdk.ChainIDKlaytn:
+		return 1 * time.Second
+	case sdk.ChainIDCelo:
+		return 10 * time.Second
+	case sdk.ChainIDNear:
+		return 2 * time.Second
+	case sdk.ChainIDMoonbeam:
+		return 24 * time.Second
+	case sdk.ChainIDTerra2:
+		return 6 * time.Second
+	case sdk.ChainIDInjective:
+		return 3 * time.Second
+	case sdk.ChainIDSui:
+		return 3 * time.Second
+	case sdk.ChainIDAptos:
+		return 4 * time.Second
+	case sdk.ChainIDArbitrum:
+		return 1066 * time.Second
+	case sdk.ChainIDOptimism:
+		return 1026 * time.Second
+	case sdk.ChainIDXpla:
+		return 5 * time.Second
+	case sdk.ChainIDBase:
+		return 1026 * time.Second
+	case sdk.ChainIDSei:
+		return 1 * time.Second
+	case sdk.ChainIDWormchain:
+		return 5 * time.Second
+	case sdk.ChainIDSepolia:
+		return 975 * time.Second
+	case sdk.ChainIDArbitrumSepolia:
+		return 1066 * time.Second
+	case sdk.ChainIDBaseSepolia:
+		return 1026 * time.Second
+	case sdk.ChainIDOptimismSepolia:
+		return 1026 * time.Second
+	case sdk.ChainIDHolesky:
+		return 975 * time.Second
+	default:
+		// The default value is the max finality time.
+		return 1066 * time.Second
+	}
+}

fly-event-processor/processor/types.go

@@ -0,0 +1,15 @@
+package processor
+
+import (
+	sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
+	"golang.org/x/net/context"
+)
+
+type Params struct {
+	TrackID string
+	VaaID   string
+	ChainID sdk.ChainID
+}
+
+// ProcessorFunc is a function to process vaa message.
+type ProcessorFunc func(context.Context, *Params) error

fly-event-processor/queue/event_sqs.go

@@ -0,0 +1,157 @@
+package queue
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"sync"
+	"time"
+
+	sqs_client "github.com/wormhole-foundation/wormhole-explorer/common/client/sqs"
+	"github.com/wormhole-foundation/wormhole-explorer/fly-event-processor/internal/metrics"
+	"go.uber.org/zap"
+)
+
+// SQSOption represents a VAA queue in SQS option function.
+type SQSOption func(*SQS)
+
+// SQS represents a VAA queue in SQS.
+type SQS struct {
+	consumer *sqs_client.Consumer
+	ch       chan ConsumerMessage
+	chSize   int
+	wg       sync.WaitGroup
+	metrics  metrics.Metrics
+	logger   *zap.Logger
+}
+
+// NewEventSqs creates a VAA queue in SQS instances.
+func NewEventSqs(consumer *sqs_client.Consumer, metrics metrics.Metrics, logger *zap.Logger, opts ...SQSOption) *SQS {
+	s := &SQS{
+		consumer: consumer,
+		chSize:   10,
+		metrics:  metrics,
+		logger:   logger.With(zap.String("queueUrl", consumer.GetQueueUrl())),
+	}
+	for _, opt := range opts {
+		opt(s)
+	}
+	s.ch = make(chan ConsumerMessage, s.chSize)
+	return s
+}
+
+// WithChannelSize allows to specify an channel size when setting a value.
+func WithChannelSize(size int) SQSOption {
+	return func(d *SQS) {
+		d.chSize = size
+	}
+}
+
+// Consume returns the channel with the received messages from SQS queue.
+func (q *SQS) Consume(ctx context.Context) <-chan ConsumerMessage {
+	go func() {
+		for {
+			messages, err := q.consumer.GetMessages(ctx)
+			if err != nil {
+				q.logger.Error("Error getting messages from SQS", zap.Error(err))
+				continue
+			}
+			q.logger.Debug("Received messages from SQS", zap.Int("count", len(messages)))
+			expiredAt := time.Now().Add(q.consumer.GetVisibilityTimeout())
+			for _, msg := range messages {
+
+				q.metrics.IncDuplicatedVaaConsumedQueue()
+				// unmarshal body to sqsEvent
+				var sqsEvent sqsEvent
+				err := json.Unmarshal([]byte(*msg.Body), &sqsEvent)
+				if err != nil {
+					q.logger.Error("Error decoding message from SQS", zap.Error(err))
+					if err = q.consumer.DeleteMessage(ctx, msg.ReceiptHandle); err != nil {
+						q.logger.Error("Error deleting message from SQS", zap.Error(err))
+					}
+					continue
+				}
+
+				var event *Event
+				err = json.Unmarshal([]byte(sqsEvent.Message), event)
+				if err != nil {
+					q.logger.Error("Error decoding message from SQS", zap.Error(err))
+					if err = q.consumer.DeleteMessage(ctx, msg.ReceiptHandle); err != nil {
+						q.logger.Error("Error deleting message from SQS", zap.Error(err))
+					}
+					continue
+				}
+
+				if event == nil {
+					q.logger.Warn("Can not handle message", zap.String("body", *msg.Body))
+					if err = q.consumer.DeleteMessage(ctx, msg.ReceiptHandle); err != nil {
+						q.logger.Error("Error deleting message from SQS", zap.Error(err))
+					}
+					continue
+				}
+
+				retry, _ := strconv.Atoi(msg.Attributes["ApproximateReceiveCount"])
+				q.wg.Add(1)
+				q.ch <- &sqsConsumerMessage{
+					id:        msg.ReceiptHandle,
+					data:      event,
+					wg:        &q.wg,
+					logger:    q.logger,
+					consumer:  q.consumer,
+					expiredAt: expiredAt,
+					retry:     uint8(retry),
+					metrics:   q.metrics,
+					ctx:       ctx,
+				}
+			}
+			q.wg.Wait()
+		}
+
+	}()
+	return q.ch
+}
+
+// Close closes all consumer resources.
+func (q *SQS) Close() {
+	close(q.ch)
+}
+
+type sqsConsumerMessage struct {
+	data      *Event
+	consumer  *sqs_client.Consumer
+	wg        *sync.WaitGroup
+	id        *string
+	logger    *zap.Logger
+	expiredAt time.Time
+	retry     uint8
+	metrics   metrics.Metrics
+	ctx       context.Context
+}
+
+func (m *sqsConsumerMessage) Data() *Event {
+	return m.data
+}
+
+func (m *sqsConsumerMessage) Done() {
+	if err := m.consumer.DeleteMessage(m.ctx, m.id); err != nil {
+		m.logger.Error("Error deleting message from SQS",
+			zap.String("vaaId", m.data.Data.VaaID),
+			zap.Bool("isExpired", m.IsExpired()),
+			zap.Time("expiredAt", m.expiredAt),
+			zap.Error(err),
+		)
+	}
+	m.wg.Done()
+}
+
+func (m *sqsConsumerMessage) Failed() {
+	m.wg.Done()
+}
+
+func (m *sqsConsumerMessage) IsExpired() bool {
+	return m.expiredAt.Before(time.Now())
+}
+
+func (m *sqsConsumerMessage) Retry() uint8 {
+	return m.retry
+}

fly-event-processor/queue/types.go

@@ -0,0 +1,43 @@
+package queue
+
+import (
+	"context"
+	"time"
+)
+
+// sqsEvent represents a event data from SQS.
+type sqsEvent struct {
+	MessageID string `json:"MessageId"`
+	Message   string `json:"Message"`
+}
+
+// Event represents a event data to be handle.
+type Event struct {
+	TrackID string       `json:"trackId"`
+	Type    string       `json:"type"`
+	Source  string       `json:"source"`
+	Data    DuplicateVaa `json:"data"`
+}
+
+type DuplicateVaa struct {
+	VaaID            string     `json:"vaaId"`
+	ChainID          uint16     `json:"chainId"`
+	Version          uint8      `json:"version"`
+	GuardianSetIndex uint32     `json:"guardianSetIndex"`
+	Vaa              []byte     `json:"vaas"`
+	Digest           string     `json:"digest"`
+	ConsistencyLevel uint8      `json:"consistencyLevel"`
+	Timestamp        *time.Time `json:"timestamp"`
+}
+
+// ConsumerMessage defition.
+type ConsumerMessage interface {
+	Retry() uint8
+	Data() *Event
+	Done()
+	Failed()
+	IsExpired() bool
+}
+
+// ConsumeFunc is a function to consume Event.
+type ConsumeFunc func(context.Context) <-chan ConsumerMessage

fly-event-processor/storage/repository.go

@@ -0,0 +1,127 @@
+package storage
+
+import (
+	"context"
+
+	commonRepo "github.com/wormhole-foundation/wormhole-explorer/common/repository"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.uber.org/zap"
+	"gopkg.in/mgo.v2/bson"
+)
+
+// Repository exposes operations over the `globalTransactions` collection.
+type Repository struct {
+	logger        *zap.Logger
+	vaas          *mongo.Collection
+	duplicateVaas *mongo.Collection
+}
+
+// New creates a new repository.
+func NewRepository(logger *zap.Logger, db *mongo.Database) *Repository {
+	r := Repository{
+		logger:        logger,
+		vaas:          db.Collection(commonRepo.Vaas),
+		duplicateVaas: db.Collection(commonRepo.DuplicateVaas),
+	}
+	return &r
+}
+
+// FindVAAById find a vaa by id.
+func (r *Repository) FindVAAById(ctx context.Context, vaaID string) (*VaaDoc, error) {
+	var vaaDoc VaaDoc
+	err := r.vaas.FindOne(ctx, bson.M{"_id": vaaID}).Decode(&vaaDoc)
+	return &vaaDoc, err
+}
+
+// FindDuplicateVAAById find a duplicate vaa by id.
+func (r *Repository) FindDuplicateVAAById(ctx context.Context, id string) (*DuplicateVaaDoc, error) {
+	var duplicateVaaDoc DuplicateVaaDoc
+	err := r.duplicateVaas.FindOne(ctx, bson.M{"_id": id}).Decode(&duplicateVaaDoc)
+	return &duplicateVaaDoc, err
+}
+
+// FindDuplicateVAAs find duplicate vaas by vaa id.
+func (r *Repository) FindDuplicateVAAs(ctx context.Context, vaaID string) ([]DuplicateVaaDoc, error) {
+	var duplicateVaaDocs []DuplicateVaaDoc
+	cursor, err := r.duplicateVaas.Find(ctx, bson.M{"vaaId": vaaID})
+	if err != nil {
+		return nil, err
+	}
+	if err = cursor.All(ctx, &duplicateVaaDocs); err != nil {
+		return nil, err
+	}
+	return duplicateVaaDocs, nil
+}
+
+// FixVAA fix a vaa by id.
+func (r *Repository) FixVAA(ctx context.Context, vaaID, duplicateID string) error {
+	// start mongo transaction
+	session, err := r.vaas.Database().Client().StartSession()
+	if err != nil {
+		return err
+	}
+
+	err = session.StartTransaction()
+	if err != nil {
+		return err
+	}
+
+	// get VAA by id
+	vaaDoc, err := r.FindVAAById(ctx, vaaID)
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	// get duplicate vaa by id
+	duplicateVaaDoc, err := r.FindDuplicateVAAById(ctx, duplicateID)
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	// create new vaa and new duplicate vaa
+	newVaa := duplicateVaaDoc.ToVaaDoc(true)
+	newDuplicateVaa, err := vaaDoc.ToDuplicateVaaDoc()
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	// remove vaa
+	_, err = r.vaas.DeleteOne(ctx, bson.M{"_id": vaaID})
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	// remove duplicate vaa
+	_, err = r.duplicateVaas.DeleteOne(ctx, bson.M{"_id": duplicateID})
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	// insert new vaa
+	_, err = r.vaas.InsertOne(ctx, newVaa)
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	// insert new duplicate vaa
+	_, err = r.duplicateVaas.InsertOne(ctx, newDuplicateVaa)
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	// commit transaction
+	err = session.CommitTransaction(ctx)
+	if err != nil {
+		session.AbortTransaction(ctx)
+		return err
+	}
+
+	return nil
+}

fly-event-processor/storage/types.go

@@ -0,0 +1,87 @@
+package storage
+
+import (
+	"time"
+
+	"github.com/wormhole-foundation/wormhole-explorer/common/domain"
+	"github.com/wormhole-foundation/wormhole/sdk/vaa"
+	sdk "github.com/wormhole-foundation/wormhole/sdk/vaa"
+)
+
+// VaaDoc represents a VAA document.
+type VaaDoc struct {
+	ID               string      `bson:"_id"`
+	Version          uint8       `bson:"version"`
+	EmitterChain     sdk.ChainID `bson:"emitterChain"`
+	EmitterAddr      string      `bson:"emitterAddr"`
+	Sequence         string      `bson:"sequence"`
+	GuardianSetIndex uint32      `bson:"guardianSetIndex"`
+	Vaa              []byte      `bson:"vaas"`
+	TxHash           string      `bson:"txHash,omitempty"`
+	OriginTxHash     *string     `bson:"_originTxHash,omitempty"` //this is temporary field for fix enconding txHash
+	Timestamp        *time.Time  `bson:"timestamp"`
+	UpdatedAt        *time.Time  `bson:"updatedAt"`
+	Digest           string      `bson:"digest"`
+	IsDuplicated     bool        `bson:"isDuplicated"`
+	DuplicatedFixed  bool        `bson:"duplicatedFixed"`
+}
+
+// DuplicateVaaDoc represents a duplicate VAA document.
+type DuplicateVaaDoc struct {
+	ID               string      `bson:"_id"`
+	VaaID            string      `bson:"vaaId"`
+	Version          uint8       `bson:"version"`
+	EmitterChain     sdk.ChainID `bson:"emitterChain"`
+	EmitterAddr      string      `bson:"emitterAddr"`
+	Sequence         string      `bson:"sequence"`
+	GuardianSetIndex uint32      `bson:"guardianSetIndex"`
+	Vaa              []byte      `bson:"vaas"`
+	Digest           string      `bson:"digest"`
+	ConsistencyLevel uint8       `bson:"consistencyLevel"`
+	TxHash           string      `bson:"txHash,omitempty"`
+	Timestamp        *time.Time  `bson:"timestamp"`
+	UpdatedAt        *time.Time  `bson:"updatedAt"`
+}
+
+func (d *DuplicateVaaDoc) ToVaaDoc(duplicatedFixed bool) *VaaDoc {
+	return &VaaDoc{
+		ID:               d.VaaID,
+		Version:          d.Version,
+		EmitterChain:     d.EmitterChain,
+		EmitterAddr:      d.EmitterAddr,
+		Sequence:         d.Sequence,
+		GuardianSetIndex: d.GuardianSetIndex,
+		Vaa:              d.Vaa,
+		Digest:           d.Digest,
+		TxHash:           d.TxHash,
+		OriginTxHash:     nil,
+		Timestamp:        d.Timestamp,
+		UpdatedAt:        d.UpdatedAt,
+		DuplicatedFixed:  duplicatedFixed,
+		IsDuplicated:     true,
+	}
+}
+
+func (v *VaaDoc) ToDuplicateVaaDoc() (*DuplicateVaaDoc, error) {
+	vaa, err := vaa.Unmarshal(v.Vaa)
+	if err != nil {
+		return nil, err
+	}
+
+	uniqueId := domain.CreateUniqueVaaID(vaa)
+	return &DuplicateVaaDoc{
+		ID:               uniqueId,
+		VaaID:            v.ID,
+		Version:          v.Version,
+		EmitterChain:     v.EmitterChain,
+		EmitterAddr:      v.EmitterAddr,
+		Sequence:         v.Sequence,
+		GuardianSetIndex: v.GuardianSetIndex,
+		Vaa:              v.Vaa,
+		Digest:           v.Digest,
+		TxHash:           v.TxHash,
+		ConsistencyLevel: vaa.ConsistencyLevel,
+		Timestamp:        v.Timestamp,
+		UpdatedAt:        v.UpdatedAt,
+	}, nil
+}

fly/builder/event.go

@@ -0,0 +1,26 @@
+package builder
+
+import (
+	"context"
+
+	"github.com/wormhole-foundation/wormhole-explorer/fly/config"
+	"github.com/wormhole-foundation/wormhole-explorer/fly/event"
+	"go.uber.org/zap"
+)
+
+func NewEventDispatcher(ctx context.Context, config *config.Configuration, logger *zap.Logger) event.EventDispatcher {
+	if config.IsLocal {
+		return event.NewNoopEventDispatcher()
+	}
+
+	awsConfig, err := NewAwsConfig(ctx, config)
+	if err != nil {
+		logger.Fatal("could not create aws config", zap.Error(err))
+	}
+
+	ed, err := event.NewSnsEventDispatcher(awsConfig, config.Aws.EventsSnsUrl)
+	if err != nil {
+		logger.Fatal("could not create sns event dispatcher", zap.Error(err))
+	}
+	return ed
+}

fly/config/config.go

@@ -77,6 +77,7 @@ type AwsConfiguration struct {
 	AwsEndpoint        string `env:"AWS_ENDPOINT"`
 	SqsUrl             string `env:"SQS_URL,required"`
 	ObservationsSqsUrl string `env:"OBSERVATIONS_SQS_URL,required"`
+	EventsSnsUrl       string `env:"EVENTS_SNS_URL,required"`
 }
 
 type Cache struct {

fly/event/noop.go

@@ -0,0 +1,13 @@
+package event
+
+import "context"
+
+type NoopEventDispatcher struct{}
+
+func NewNoopEventDispatcher() *NoopEventDispatcher {
+	return &NoopEventDispatcher{}
+}
+
+func (n *NoopEventDispatcher) NewDuplicateVaa(context.Context, DuplicateVaa) error {
+	return nil
+}

fly/event/sns.go

@@ -0,0 +1,44 @@
+package event
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	aws_sns "github.com/aws/aws-sdk-go-v2/service/sns"
+	"github.com/wormhole-foundation/wormhole-explorer/fly/internal/track"
+)
+
+type SnsEventDispatcher struct {
+	api *aws_sns.Client
+	url string
+}
+
+func NewSnsEventDispatcher(awsConfig aws.Config, url string) (*SnsEventDispatcher, error) {
+	return &SnsEventDispatcher{
+		api: aws_sns.NewFromConfig(awsConfig),
+		url: url,
+	}, nil
+}
+
+func (s *SnsEventDispatcher) NewDuplicateVaa(ctx context.Context, e DuplicateVaa) error {
+	body, err := json.Marshal(event{
+		TrackID: track.GetTrackIDForDuplicatedVAA(e.VaaID),
+		Type:    "duplicated-vaa",
+		Source:  "fly",
+		Data:    e,
+	})
+	if err != nil {
+		return err
+	}
+	groupID := fmt.Sprintf("%s-%s", e.VaaID, e.Digest)
+	_, err = s.api.Publish(ctx,
+		&aws_sns.PublishInput{
+			MessageGroupId:         aws.String(groupID),
+			MessageDeduplicationId: aws.String(groupID),
+			Message:                aws.String(string(body)),
+			TopicArn:               aws.String(s.url),
+		})
+	return err
+}

fly/event/types.go

@@ -0,0 +1,28 @@
+package event
+
+import (
+	"context"
+	"time"
+)
+
+type DuplicateVaa struct {
+	VaaID            string     `json:"vaaId"`
+	ChainID          uint16     `json:"chainId"`
+	Version          uint8      `json:"version"`
+	GuardianSetIndex uint32     `json:"guardianSetIndex"`
+	Vaa              []byte     `json:"vaas"`
+	Digest           string     `json:"digest"`
+	ConsistencyLevel uint8      `json:"consistencyLevel"`
+	Timestamp        *time.Time `json:"timestamp"`
+}
+
+type event struct {
+	TrackID string `json:"trackId"`
+	Type    string `json:"type"`
+	Source  string `json:"source"`
+	Data    any    `json:"data"`
+}
+
+type EventDispatcher interface {
+	NewDuplicateVaa(ctx context.Context, e DuplicateVaa) error
+}

fly/internal/track/trackid.go

@@ -11,3 +11,8 @@ func GetTrackID(vaaID string) string {
 	uuid := uuid.New()
 	return fmt.Sprintf("gossip-signed-vaa-%s-%s", vaaID, uuid.String())
 }
+
+func GetTrackIDForDuplicatedVAA(vaaID string) string {
+	uuid := uuid.New()
+	return fmt.Sprintf("fly-duplicated-vaa-%s-%s", vaaID, uuid.String())
+}

fly/main.go

@@ -99,7 +99,9 @@ func main() {
 	if err != nil {
 		logger.Fatal("could not create tx hash store", zap.Error(err))
 	}
-	repository := storage.NewRepository(alertClient, metrics, db.Database, producerFunc, txHashStore, logger)
+	eventDispatcher := builder.NewEventDispatcher(rootCtx, cfg, logger)
+
+	repository := storage.NewRepository(alertClient, metrics, db.Database, producerFunc, txHashStore, eventDispatcher, logger)
 
 	vaaNonPythDedup, err := builder.NewDeduplicator("vaas-dedup", cfg.VaasDedup, logger)
 	if err != nil {

fly/storage/repository.go

@@ -15,6 +15,7 @@ import (
 	"github.com/wormhole-foundation/wormhole-explorer/common/events"
 	"github.com/wormhole-foundation/wormhole-explorer/common/repository"
 	"github.com/wormhole-foundation/wormhole-explorer/common/utils"
+	"github.com/wormhole-foundation/wormhole-explorer/fly/event"
 	flyAlert "github.com/wormhole-foundation/wormhole-explorer/fly/internal/alert"
 	"github.com/wormhole-foundation/wormhole-explorer/fly/internal/metrics"
 	"github.com/wormhole-foundation/wormhole-explorer/fly/internal/track"
@@ -30,13 +31,14 @@ import (
 
 // TODO separate and maybe share between fly and web
 type Repository struct {
-	alertClient alert.AlertClient
-	metrics     metrics.Metrics
-	db          *mongo.Database
-	afterUpdate producer.PushFunc
-	txHashStore txhash.TxHashStore
-	log         *zap.Logger
-	collections struct {
+	alertClient     alert.AlertClient
+	metrics         metrics.Metrics
+	db              *mongo.Database
+	afterUpdate     producer.PushFunc
+	txHashStore     txhash.TxHashStore
+	eventDispatcher event.EventDispatcher
+	log             *zap.Logger
+	collections     struct {
 		vaas           *mongo.Collection
 		heartbeats     *mongo.Collection
 		observations   *mongo.Collection
@@ -53,8 +55,9 @@ func NewRepository(alertService alert.AlertClient, metrics metrics.Metrics,
 	db *mongo.Database,
 	vaaTopicFunc producer.PushFunc,
 	txHashStore txhash.TxHashStore,
+	eventDispatcher event.EventDispatcher,
 	log *zap.Logger) *Repository {
-	return &Repository{alertService, metrics, db, vaaTopicFunc, txHashStore, log, struct {
+	return &Repository{alertService, metrics, db, vaaTopicFunc, txHashStore, eventDispatcher, log, struct {
 		vaas           *mongo.Collection
 		heartbeats     *mongo.Collection
 		observations   *mongo.Collection
@@ -514,7 +517,20 @@ func (s *Repository) UpsertDuplicateVaa(ctx context.Context, v *vaa.VAA, seriali
 
 	// send signedvaa event to topic.
 	if s.isNewRecord(result) {
-		return s.notifyNewVaa(ctx, v, serializedVaa, duplicateVaaDoc.TxHash)
+		err := s.notifyNewVaa(ctx, v, serializedVaa, duplicateVaaDoc.TxHash)
+		if err != nil {
+			return err
+		}
+		return s.eventDispatcher.NewDuplicateVaa(ctx, event.DuplicateVaa{
+			VaaID:            v.MessageID(),
+			ChainID:          uint16(v.EmitterChain),
+			Version:          v.Version,
+			GuardianSetIndex: v.GuardianSetIndex,
+			Vaa:              serializedVaa,
+			Digest:           utils.NormalizeHex(v.HexDigest()),
+			ConsistencyLevel: v.ConsistencyLevel,
+			Timestamp:        &v.Timestamp,
+		})
 	}
 
 	return nil

go.work

@@ -12,4 +12,5 @@ use (
 	./spy
 	./tx-tracker
 	./notional
+	./fly-event-processor
 )

pipeline/pipeline/publisher.go

@@ -49,7 +49,8 @@ func (p *Publisher) Publish(ctx context.Context, e *watcher.Event) {
 		TxHash:           e.TxHash,
 		Version:          e.Version,
 		Revision:         e.Revision,
-		Hash:             e.Hash,
+		Digest:           e.Digest,
+		Overwrite:        e.DuplicatedFixed,
 	}
 
 	// In some scenarios the fly component that inserts the VAA documents does not have the txhash field available,

pipeline/topic/topic.go

@@ -19,7 +19,8 @@ type Event struct {
 	TxHash           string     `json:"txHash"`
 	Version          uint16     `json:"version"`
 	Revision         uint16     `json:"revision"`
-	Hash             []byte     `json:"hash"`
+	Digest           string     `json:"digest"`
+	Overwrite        bool       `json:"overwrite"`
 }
 
 // PushFunc is a function to push VAAEvent.

pipeline/watcher/watcher.go

@@ -50,8 +50,9 @@ type Event struct {
 	TxHash           string     `bson:"txHash"`
 	Version          uint16     `bson:"version"`
 	Revision         uint16     `bson:"revision"`
-	Hash             []byte     `bson:"hash"`
+	Digest           string     `bson:"digest"`
 	IsDuplicated     bool       `bson:"isDuplicated"`
+	DuplicatedFixed  bool       `bson:"duplicatedFixed"`
 }
 
 const queryTemplate = `

tx-tracker/consumer/consumer.go

@@ -114,7 +114,7 @@ func (c *Consumer) processSourceTx(ctx context.Context, msg queue.ConsumerMessag
 		Vaa:         event.Vaa,
 		IsVaaSigned: event.IsVaaSigned,
 		Metrics:     c.metrics,
-		Overwrite:   false, // avoid processing the same transaction twice
+		Overwrite:   event.Overwrite, // avoid processing the same transaction twice
 		Source:      event.Source,
 	}
 	_, err := ProcessSourceTx(ctx, c.logger, c.rpcpool, c.wormchainRpcPool, c.repository, &p, c.p2pNetwork)

tx-tracker/consumer/repository.go

@@ -159,270 +159,11 @@ func (r *Repository) AlreadyProcessed(ctx context.Context, vaaId string) (bool,
 	}
 }
 
-// CountDocumentsByTimeRange returns the number of documents that match the given time range.
-func (r *Repository) CountDocumentsByTimeRange(
-	ctx context.Context,
-	timeAfter time.Time,
-	timeBefore time.Time,
-) (uint64, error) {
-
-	// Build the aggregation pipeline
-	var pipeline mongo.Pipeline
-	{
-		// filter by time range
-		pipeline = append(pipeline, bson.D{
-			{"$match", bson.D{
-				{"timestamp", bson.D{{"$gte", timeAfter}}},
-			}},
-		})
-		pipeline = append(pipeline, bson.D{
-			{"$match", bson.D{
-				{"timestamp", bson.D{{"$lte", timeBefore}}},
-			}},
-		})
-
-		// Count the number of results
-		pipeline = append(pipeline, bson.D{
-			{"$count", "numDocuments"},
-		})
-	}
-
-	// Execute the aggregation pipeline
-	cur, err := r.vaas.Aggregate(ctx, pipeline)
-	if err != nil {
-		r.logger.Error("failed execute aggregation pipeline", zap.Error(err))
-		return 0, err
-	}
-
-	// Read results from cursor
-	var results []struct {
-		NumDocuments uint64 `bson:"numDocuments"`
-	}
-	err = cur.All(ctx, &results)
-	if err != nil {
-		r.logger.Error("failed to decode cursor", zap.Error(err))
-		return 0, err
-	}
-	if len(results) == 0 {
-		return 0, nil
-	}
-	if len(results) > 1 {
-		r.logger.Error("too many results", zap.Int("numResults", len(results)))
-		return 0, err
-	}
-
-	return results[0].NumDocuments, nil
-}
-
-// CountIncompleteDocuments returns the number of documents that have destTx data, but don't have sourceTx data.
-func (r *Repository) CountIncompleteDocuments(ctx context.Context) (uint64, error) {
-
-	// Build the aggregation pipeline
-	var pipeline mongo.Pipeline
-	{
-		// Look up transactions that either:
-		// 1. have not been processed
-		// 2. have been processed, but encountered an internal error
-		pipeline = append(pipeline, bson.D{
-			{"$match", bson.D{
-				{"$or", bson.A{
-					bson.D{{"originTx", bson.D{{"$exists", false}}}},
-					bson.D{{"originTx.status", bson.M{"$eq": domain.SourceTxStatusInternalError}}},
-				}},
-			}},
-		})
-
-		// Count the number of results
-		pipeline = append(pipeline, bson.D{
-			{"$count", "numDocuments"},
-		})
-	}
-
-	// Execute the aggregation pipeline
-	cur, err := r.globalTransactions.Aggregate(ctx, pipeline)
-	if err != nil {
-		r.logger.Error("failed execute aggregation pipeline", zap.Error(err))
-		return 0, err
-	}
-
-	// Read results from cursor
-	var results []struct {
-		NumDocuments uint64 `bson:"numDocuments"`
-	}
-	err = cur.All(ctx, &results)
-	if err != nil {
-		r.logger.Error("failed to decode cursor", zap.Error(err))
-		return 0, err
-	}
-	if len(results) == 0 {
-		return 0, nil
-	}
-	if len(results) > 1 {
-		r.logger.Error("too many results", zap.Int("numResults", len(results)))
-		return 0, err
-	}
-
-	return results[0].NumDocuments, nil
-}
-
 type GlobalTransaction struct {
 	Id   string       `bson:"_id"`
 	Vaas []vaa.VaaDoc `bson:"vaas"`
 }
 
-// GetDocumentsByTimeRange iterates through documents within a specified time range.
-func (r *Repository) GetDocumentsByTimeRange(
-	ctx context.Context,
-	lastId string,
-	lastTimestamp *time.Time,
-	limit uint,
-	timeAfter time.Time,
-	timeBefore time.Time,
-) ([]GlobalTransaction, error) {
-
-	// Build the aggregation pipeline
-	var pipeline mongo.Pipeline
-	{
-		// Specify sorting criteria
-		pipeline = append(pipeline, bson.D{
-			{"$sort", bson.D{
-				bson.E{"timestamp", -1},
-				bson.E{"_id", 1},
-			}},
-		})
-
-		// filter out already processed documents
-		//
-		// We use the timestap field as a pagination cursor
-		if lastTimestamp != nil {
-			pipeline = append(pipeline, bson.D{
-				{"$match", bson.D{
-					{"$or", bson.A{
-						bson.D{{"timestamp", bson.M{"$lt": *lastTimestamp}}},
-						bson.D{{"$and", bson.A{
-							bson.D{{"timestamp", bson.M{"$eq": *lastTimestamp}}},
-							bson.D{{"_id", bson.M{"$gt": lastId}}},
-						}}},
-					}},
-				}},
-			})
-		}
-
-		// filter by time range
-		pipeline = append(pipeline, bson.D{
-			{"$match", bson.D{
-				{"timestamp", bson.D{{"$gte", timeAfter}}},
-			}},
-		})
-		pipeline = append(pipeline, bson.D{
-			{"$match", bson.D{
-				{"timestamp", bson.D{{"$lte", timeBefore}}},
-			}},
-		})
-
-		// Limit size of results
-		pipeline = append(pipeline, bson.D{
-			{"$limit", limit},
-		})
-	}
-
-	// Execute the aggregation pipeline
-	cur, err := r.vaas.Aggregate(ctx, pipeline)
-	if err != nil {
-		r.logger.Error("failed execute aggregation pipeline", zap.Error(err))
-		return nil, errors.WithStack(err)
-	}
-
-	// Read results from cursor
-	var documents []vaa.VaaDoc
-	err = cur.All(ctx, &documents)
-	if err != nil {
-		r.logger.Error("failed to decode cursor", zap.Error(err))
-		return nil, errors.WithStack(err)
-	}
-
-	// Build the result
-	var globalTransactions []GlobalTransaction
-	for i := range documents {
-		globalTransaction := GlobalTransaction{
-			Id:   documents[i].ID,
-			Vaas: []vaa.VaaDoc{documents[i]},
-		}
-		globalTransactions = append(globalTransactions, globalTransaction)
-	}
-
-	return globalTransactions, nil
-}
-
-// GetIncompleteDocuments gets a batch of VAA IDs from the database.
-func (r *Repository) GetIncompleteDocuments(
-	ctx context.Context,
-	lastId string,
-	lastTimestamp *time.Time,
-	limit uint,
-) ([]GlobalTransaction, error) {
-
-	// Build the aggregation pipeline
-	var pipeline mongo.Pipeline
-	{
-		// Specify sorting criteria
-		pipeline = append(pipeline, bson.D{
-			{"$sort", bson.D{bson.E{"_id", 1}}},
-		})
-
-		// filter out already processed documents
-		//
-		// We use the _id field as a pagination cursor
-		pipeline = append(pipeline, bson.D{
-			{"$match", bson.D{{"_id", bson.M{"$gt": lastId}}}},
-		})
-
-		// Look up transactions that either:
-		// 1. have not been processed
-		// 2. have been processed, but encountered an internal error
-		pipeline = append(pipeline, bson.D{
-			{"$match", bson.D{
-				{"$or", bson.A{
-					bson.D{{"originTx", bson.D{{"$exists", false}}}},
-					bson.D{{"originTx.status", bson.M{"$eq": domain.SourceTxStatusInternalError}}},
-				}},
-			}},
-		})
-
-		// Left join on the VAA collection
-		pipeline = append(pipeline, bson.D{
-			{"$lookup", bson.D{
-				{"from", "vaas"},
-				{"localField", "_id"},
-				{"foreignField", "_id"},
-				{"as", "vaas"},
-			}},
-		})
-
-		// Limit size of results
-		pipeline = append(pipeline, bson.D{
-			{"$limit", limit},
-		})
-	}
-
-	// Execute the aggregation pipeline
-	cur, err := r.globalTransactions.Aggregate(ctx, pipeline)
-	if err != nil {
-		r.logger.Error("failed execute aggregation pipeline", zap.Error(err))
-		return nil, errors.WithStack(err)
-	}
-
-	// Read results from cursor
-	var documents []GlobalTransaction
-	err = cur.All(ctx, &documents)
-	if err != nil {
-		r.logger.Error("failed to decode cursor", zap.Error(err))
-		return nil, errors.WithStack(err)
-	}
-
-	return documents, nil
-}
-
 // VaaIdTxHash represents a vaaIdTxHash document.
 type VaaIdTxHash struct {
 	TxHash string `bson:"txHash"`

tx-tracker/queue/converter.go

@@ -25,6 +25,7 @@ type VaaEvent struct {
 	TxHash           string      `json:"txHash"`
 	Version          uint16      `json:"version"`
 	Revision         uint16      `json:"revision"`
+	Overwrite        bool        `json:"overwrite"`
 }
 
 // VaaConverter converts a message from a VAAEvent.
@@ -49,6 +50,7 @@ func NewVaaConverter(log *zap.Logger) ConverterFunc {
 			Vaa:            vaaEvent.Vaa,
 			IsVaaSigned:    true,
 			TxHash:         vaaEvent.TxHash,
+			Overwrite:      vaaEvent.Overwrite,
 		}, nil
 	}
 }

tx-tracker/queue/types.go

@@ -51,6 +51,7 @@ type Event struct {
 	Vaa            []byte
 	IsVaaSigned    bool
 	Attributes     any
+	Overwrite      bool
 }
 
 func GetAttributes[T EventAttributes](e *Event) (T, bool) {