Merge ac862cde08 into ce1b7707fb

* add 2 new tasks for collecting chain activity every day and hour

* making progress

* change query 2

* add query by month and year

* changes on task

* more changes

* change to 1d

* add 1d

* fix query

* adjust queryies

* change the way the query is executed

* changes on query

* making more progress

* fix per year query

* add a second group of tasks for downsampling

* add app_id

* update swagger docs

* optimize new tasks

* fix W

* fix W

* start using the new measurement

* change endpoint signature

* update endpoint name

* fix indents

* code review changes

* remove unnecessary break

analytics/scripts/chain_activity_emitter_1d.flux

@@ -0,0 +1,40 @@
+import "date"
+
+
+runTask = (start,stop,srcBucket,destBucket,destMeasurement) => {
+
+    data = from(bucket: srcBucket)
+                 	|> range(start: start,stop: stop)
+                 	|> filter(fn: (r) => r._measurement == "vaa_volume_v2" and r.version == "v2")
+                 	|> filter(fn: (r) => r._field == "volume" and r._value > 0)
+               		|> drop(columns:["destination_chain","app_id","token_chain","token_address","version","_measurement","_time"])
+               		|> rename(columns: {_start: "_time"})
+                 	|> group(columns: ["emitter_chain","_time"])
+
+    vols =	data
+    			|> sum(column: "_value")
+    			|> set(key: "_field", value: "volume")
+    			|> set(key: "to", value: string(v:stop))
+    			|> set(key: "_measurement", value: destMeasurement)
+    			|> to(bucket: destBucket)
+
+    return data
+    		    |> count(column: "_value")
+    		    |> set(key: "_field", value: "count")
+    		    |> set(key: "to", value: string(v:stop))
+    		    |> set(key: "_measurement", value: destMeasurement)
+    		    |> to(bucket: destBucket)
+}
+
+bucketInfinite = "wormscan"
+destMeasurement = "emitter_chain_activity_1d"
+
+stop = date.truncate(t: now(),unit: 1d)
+start = date.sub(d: 1d, from: stop)
+
+option task = {
+    name: "calculate chain activity per emitter every day",
+    every: 1d,
+}
+
+runTask(start:start, stop: stop, srcBucket: bucketInfinite, destBucket: bucketInfinite, destMeasurement: destMeasurement)

analytics/scripts/chain_activity_emitter_1h.flux

@@ -0,0 +1,40 @@
+import "date"
+
+
+runTask = (start,stop,srcBucket,destBucket,destMeasurement) => {
+
+    data = from(bucket: srcBucket)
+                 	|> range(start: start,stop: stop)
+                 	|> filter(fn: (r) => r._measurement == "vaa_volume_v2" and r.version == "v2")
+                 	|> filter(fn: (r) => r._field == "volume" and r._value > 0)
+               		|> drop(columns:["destination_chain","app_id","token_chain","token_address","version","_measurement","_time"])
+               		|> rename(columns: {_start: "_time"})
+                 	|> group(columns: ["emitter_chain","_time"])
+
+    vols =	data
+    			|> sum(column: "_value")
+    			|> set(key: "_field", value: "volume")
+    			|> set(key: "to", value: string(v:stop))
+    			|> set(key: "_measurement", value: destMeasurement)
+    			|> to(bucket: destBucket)
+
+    return data
+    		    |> count(column: "_value")
+    		    |> set(key: "_field", value: "count")
+    		    |> set(key: "to", value: string(v:stop))
+    		    |> set(key: "_measurement", value: destMeasurement)
+    		    |> to(bucket: destBucket)
+}
+
+bucketInfinite = "wormscan"
+destMeasurement = "emitter_chain_activity_1h"
+
+stop = date.truncate(t: now(),unit: 1h)
+start = date.sub(d: 1h, from: stop)
+
+option task = {
+    name: "calculate chain activity per emitter every hour",
+    every: 1h,
+}
+
+runTask(start:start, stop: stop, srcBucket: bucketInfinite, destBucket: bucketInfinite, destMeasurement: destMeasurement)

analytics/scripts/chain_activity_emitter_dest_1d.flux

@@ -0,0 +1,37 @@
+import "date"
+
+runTask = (start,stop,srcBucket,destBucket,destMeasurement) => {
+        data = from(bucket: srcBucket)
+      		    |> range(start: start,stop: stop)
+      		    |> filter(fn: (r) => r._measurement == "vaa_volume_v2" and r._field == "volume")
+      		    |> group(columns: ["emitter_chain", "destination_chain", "app_id"])
+
+        data
+    		|> sum(column: "_value")
+    		|> set(key: "_field", value: "volume")
+    		|> map(fn: (r) => ({ r with _time: start }))
+    		|> set(key: "to", value: string(v:date.add(d: 1d, to: start)))
+    		|> set(key: "_measurement", value: destMeasurement)
+    		|> to(bucket: destBucket)
+
+        return data
+    		        |> count(column: "_value")
+    		        |> set(key: "_field", value: "count")
+    		        |> map(fn: (r) => ({ r with _time: start }))
+    		        |> set(key: "to", value: string(v:date.add(d: 1d, to: start)))
+    		        |> set(key: "_measurement", value: destMeasurement)
+    		        |> to(bucket: destBucket)
+}
+
+
+bucketInfinite = "wormscan"
+destMeasurement = "chain_activity_1d"
+stop = date.truncate(t: now(),unit: 24h)
+start = date.sub(d: 1d, from: stop)
+
+option task = {
+    name: "calculate chain activity every day",
+    every: 1d,
+}
+
+runTask(start:start, stop: stop, srcBucket: bucketInfinite, destBucket: bucketInfinite, destMeasurement: destMeasurement)

analytics/scripts/chain_activity_emitter_dest_1h.flux

@@ -0,0 +1,40 @@
+import "date"
+
+
+runTask = (start,stop,srcBucket,destBucket,destMeasurement) => {
+
+    data = from(bucket: srcBucket)
+  		    |> range(start: start,stop: stop)
+  		    |> filter(fn: (r) => r._measurement == "vaa_volume_v2" and r._field == "volume")
+  		    |> group(columns: ["emitter_chain", "destination_chain", "app_id"])
+
+    data
+		|> sum(column: "_value")
+		|> set(key: "_field", value: "volume")
+		|> map(fn: (r) => ({ r with _time: start }))
+		|> set(key: "to", value: string(v:date.add(d: 1h, to: start)))
+		|> set(key: "_measurement", value: destMeasurement)
+		|> to(bucket: destBucket)
+
+    return data
+		        |> count(column: "_value")
+		        |> set(key: "_field", value: "count")
+		        |> map(fn: (r) => ({ r with _time: start }))
+		        |> set(key: "to", value: string(v:date.add(d: 1h, to: start)))
+		        |> set(key: "_measurement", value: destMeasurement)
+		        |> to(bucket: destBucket)
+}
+
+
+bucketInfinite = "wormscan"
+destMeasurement = "chain_activity_1h"
+
+stop = date.truncate(t: now(),unit: 1h)
+start = date.sub(d: 1h, from: stop)
+
+option task = {
+    name: "calculate chain activity every hour",
+    every: 1h,
+}
+
+runTask(start:start, stop: stop, srcBucket: bucketInfinite, destBucket: bucketInfinite, destMeasurement: destMeasurement)

api/docs/docs.go

@@ -1760,6 +1760,73 @@ const docTemplate = `{
                 }
             }
         },
+        "/api/v1/x-chain-activity/tops": {
+            "get": {
+                "description": "Search, for a specific period of time, the number of transactions and the volume.",
+                "tags": [
+                    "wormholescan"
+                ],
+                "operationId": "x-chain-activity-tops",
+                "parameters": [
+                    {
+                        "type": "string",
+                        "description": "Time span, supported values: 1d, 1mo and 1y",
+                        "name": "timespan",
+                        "in": "query",
+                        "required": true
+                    },
+                    {
+                        "type": "string",
+                        "description": "From date, supported format 2006-01-02T15:04:05Z07:00",
+                        "name": "from",
+                        "in": "query",
+                        "required": true
+                    },
+                    {
+                        "type": "string",
+                        "description": "To date, supported format 2006-01-02T15:04:05Z07:00",
+                        "name": "to",
+                        "in": "query",
+                        "required": true
+                    },
+                    {
+                        "type": "string",
+                        "description": "Search by appId",
+                        "name": "appId",
+                        "in": "query"
+                    },
+                    {
+                        "type": "string",
+                        "description": "Search by sourceChain",
+                        "name": "sourceChain",
+                        "in": "query"
+                    },
+                    {
+                        "type": "string",
+                        "description": "Search by targetChain",
+                        "name": "targetChain",
+                        "in": "query"
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "schema": {
+                            "type": "array",
+                            "items": {
+                                "$ref": "#/definitions/transactions.ChainActivityTopResult"
+                            }
+                        }
+                    },
+                    "400": {
+                        "description": "Bad Request"
+                    },
+                    "500": {
+                        "description": "Internal Server Error"
+                    }
+                }
+            }
+        },
         "/swagger.json": {
             "get": {
                 "description": "Returns the swagger specification for this API.",
@@ -3325,6 +3392,29 @@ const docTemplate = `{
                 }
             }
         },
+        "transactions.ChainActivityTopResult": {
+            "type": "object",
+            "properties": {
+                "count": {
+                    "type": "integer"
+                },
+                "destination_chain": {
+                    "type": "string"
+                },
+                "emitter_chain": {
+                    "type": "string"
+                },
+                "from": {
+                    "type": "string"
+                },
+                "to": {
+                    "type": "string"
+                },
+                "volume": {
+                    "type": "integer"
+                }
+            }
+        },
         "transactions.ChainPair": {
             "type": "object",
             "properties": {
@@ -3591,7 +3681,6 @@ const docTemplate = `{
                 14,
                 15,
                 16,
-                17,
                 18,
                 19,
                 20,
@@ -3599,24 +3688,35 @@ const docTemplate = `{
                 22,
                 23,
                 24,
+                25,
                 26,
                 28,
                 29,
                 30,
                 32,
+                33,
                 34,
                 35,
+                36,
+                37,
+                38,
+                39,
                 3104,
                 4000,
                 4001,
                 4002,
                 4003,
                 4004,
+                4005,
+                4006,
+                4007,
+                4008,
                 10002,
                 10003,
                 10004,
                 10005,
-                10006
+                10006,
+                10007
             ],
             "x-enum-varnames": [
                 "ChainIDUnset",
@@ -3636,7 +3736,6 @@ const docTemplate = `{
                 "ChainIDCelo",
                 "ChainIDNear",
                 "ChainIDMoonbeam",
-                "ChainIDNeon",
                 "ChainIDTerra2",
                 "ChainIDInjective",
                 "ChainIDOsmosis",
@@ -3644,24 +3743,35 @@ const docTemplate = `{
                 "ChainIDAptos",
                 "ChainIDArbitrum",
                 "ChainIDOptimism",
+                "ChainIDGnosis",
                 "ChainIDPythNet",
                 "ChainIDXpla",
                 "ChainIDBtc",
                 "ChainIDBase",
                 "ChainIDSei",
+                "ChainIDRootstock",
                 "ChainIDScroll",
                 "ChainIDMantle",
+                "ChainIDBlast",
+                "ChainIDXLayer",
+                "ChainIDLinea",
+                "ChainIDBerachain",
                 "ChainIDWormchain",
                 "ChainIDCosmoshub",
                 "ChainIDEvmos",
                 "ChainIDKujira",
                 "ChainIDNeutron",
                 "ChainIDCelestia",
+                "ChainIDStargaze",
+                "ChainIDSeda",
+                "ChainIDDymension",
+                "ChainIDProvenance",
                 "ChainIDSepolia",
                 "ChainIDArbitrumSepolia",
                 "ChainIDBaseSepolia",
                 "ChainIDOptimismSepolia",
-                "ChainIDHolesky"
+                "ChainIDHolesky",
+                "ChainIDPolygonSepolia"
             ]
         },
         "vaa.VaaDoc": {

api/docs/swagger.json

@@ -1753,6 +1753,73 @@
                 }
             }
         },
+        "/api/v1/x-chain-activity/tops": {
+            "get": {
+                "description": "Search, for a specific period of time, the number of transactions and the volume.",
+                "tags": [
+                    "wormholescan"
+                ],
+                "operationId": "x-chain-activity-tops",
+                "parameters": [
+                    {
+                        "type": "string",
+                        "description": "Time span, supported values: 1d, 1mo and 1y",
+                        "name": "timespan",
+                        "in": "query",
+                        "required": true
+                    },
+                    {
+                        "type": "string",
+                        "description": "From date, supported format 2006-01-02T15:04:05Z07:00",
+                        "name": "from",
+                        "in": "query",
+                        "required": true
+                    },
+                    {
+                        "type": "string",
+                        "description": "To date, supported format 2006-01-02T15:04:05Z07:00",
+                        "name": "to",
+                        "in": "query",
+                        "required": true
+                    },
+                    {
+                        "type": "string",
+                        "description": "Search by appId",
+                        "name": "appId",
+                        "in": "query"
+                    },
+                    {
+                        "type": "string",
+                        "description": "Search by sourceChain",
+                        "name": "sourceChain",
+                        "in": "query"
+                    },
+                    {
+                        "type": "string",
+                        "description": "Search by targetChain",
+                        "name": "targetChain",
+                        "in": "query"
+                    }
+                ],
+                "responses": {
+                    "200": {
+                        "description": "OK",
+                        "schema": {
+                            "type": "array",
+                            "items": {
+                                "$ref": "#/definitions/transactions.ChainActivityTopResult"
+                            }
+                        }
+                    },
+                    "400": {
+                        "description": "Bad Request"
+                    },
+                    "500": {
+                        "description": "Internal Server Error"
+                    }
+                }
+            }
+        },
         "/swagger.json": {
             "get": {
                 "description": "Returns the swagger specification for this API.",
@@ -3318,6 +3385,29 @@
                 }
             }
         },
+        "transactions.ChainActivityTopResult": {
+            "type": "object",
+            "properties": {
+                "count": {
+                    "type": "integer"
+                },
+                "destination_chain": {
+                    "type": "string"
+                },
+                "emitter_chain": {
+                    "type": "string"
+                },
+                "from": {
+                    "type": "string"
+                },
+                "to": {
+                    "type": "string"
+                },
+                "volume": {
+                    "type": "integer"
+                }
+            }
+        },
         "transactions.ChainPair": {
             "type": "object",
             "properties": {
@@ -3584,7 +3674,6 @@
                 14,
                 15,
                 16,
-                17,
                 18,
                 19,
                 20,
@@ -3592,24 +3681,35 @@
                 22,
                 23,
                 24,
+                25,
                 26,
                 28,
                 29,
                 30,
                 32,
+                33,
                 34,
                 35,
+                36,
+                37,
+                38,
+                39,
                 3104,
                 4000,
                 4001,
                 4002,
                 4003,
                 4004,
+                4005,
+                4006,
+                4007,
+                4008,
                 10002,
                 10003,
                 10004,
                 10005,
-                10006
+                10006,
+                10007
             ],
             "x-enum-varnames": [
                 "ChainIDUnset",
@@ -3629,7 +3729,6 @@
                 "ChainIDCelo",
                 "ChainIDNear",
                 "ChainIDMoonbeam",
-                "ChainIDNeon",
                 "ChainIDTerra2",
                 "ChainIDInjective",
                 "ChainIDOsmosis",
@@ -3637,24 +3736,35 @@
                 "ChainIDAptos",
                 "ChainIDArbitrum",
                 "ChainIDOptimism",
+                "ChainIDGnosis",
                 "ChainIDPythNet",
                 "ChainIDXpla",
                 "ChainIDBtc",
                 "ChainIDBase",
                 "ChainIDSei",
+                "ChainIDRootstock",
                 "ChainIDScroll",
                 "ChainIDMantle",
+                "ChainIDBlast",
+                "ChainIDXLayer",
+                "ChainIDLinea",
+                "ChainIDBerachain",
                 "ChainIDWormchain",
                 "ChainIDCosmoshub",
                 "ChainIDEvmos",
                 "ChainIDKujira",
                 "ChainIDNeutron",
                 "ChainIDCelestia",
+                "ChainIDStargaze",
+                "ChainIDSeda",
+                "ChainIDDymension",
+                "ChainIDProvenance",
                 "ChainIDSepolia",
                 "ChainIDArbitrumSepolia",
                 "ChainIDBaseSepolia",
                 "ChainIDOptimismSepolia",
-                "ChainIDHolesky"
+                "ChainIDHolesky",
+                "ChainIDPolygonSepolia"
             ]
         },
         "vaa.VaaDoc": {

api/docs/swagger.yaml

@@ -824,6 +824,21 @@ definitions:
           $ref: '#/definitions/transactions.Tx'
         type: array
     type: object
+  transactions.ChainActivityTopResult:
+    properties:
+      count:
+        type: integer
+      destination_chain:
+        type: string
+      emitter_chain:
+        type: string
+      from:
+        type: string
+      to:
+        type: string
+      volume:
+        type: integer
+    type: object
   transactions.ChainPair:
     properties:
       destinationChain:
@@ -1012,7 +1027,6 @@ definitions:
     - 14
     - 15
     - 16
-    - 17
     - 18
     - 19
     - 20
@@ -1020,24 +1034,35 @@ definitions:
     - 22
     - 23
     - 24
+    - 25
     - 26
     - 28
     - 29
     - 30
     - 32
+    - 33
     - 34
     - 35
+    - 36
+    - 37
+    - 38
+    - 39
     - 3104
     - 4000
     - 4001
     - 4002
     - 4003
     - 4004
+    - 4005
+    - 4006
+    - 4007
+    - 4008
     - 10002
     - 10003
     - 10004
     - 10005
     - 10006
+    - 10007
     type: integer
     x-enum-varnames:
     - ChainIDUnset
@@ -1057,7 +1082,6 @@ definitions:
     - ChainIDCelo
     - ChainIDNear
     - ChainIDMoonbeam
-    - ChainIDNeon
     - ChainIDTerra2
     - ChainIDInjective
     - ChainIDOsmosis
@@ -1065,24 +1089,35 @@ definitions:
     - ChainIDAptos
     - ChainIDArbitrum
     - ChainIDOptimism
+    - ChainIDGnosis
     - ChainIDPythNet
     - ChainIDXpla
     - ChainIDBtc
     - ChainIDBase
     - ChainIDSei
+    - ChainIDRootstock
     - ChainIDScroll
     - ChainIDMantle
+    - ChainIDBlast
+    - ChainIDXLayer
+    - ChainIDLinea
+    - ChainIDBerachain
     - ChainIDWormchain
     - ChainIDCosmoshub
     - ChainIDEvmos
     - ChainIDKujira
     - ChainIDNeutron
     - ChainIDCelestia
+    - ChainIDStargaze
+    - ChainIDSeda
+    - ChainIDDymension
+    - ChainIDProvenance
     - ChainIDSepolia
     - ChainIDArbitrumSepolia
     - ChainIDBaseSepolia
     - ChainIDOptimismSepolia
     - ChainIDHolesky
+    - ChainIDPolygonSepolia
   vaa.VaaDoc:
     properties:
       appId:
@@ -2326,6 +2361,52 @@ paths:
           description: Internal Server Error
       tags:
       - wormholescan
+  /api/v1/x-chain-activity/tops:
+    get:
+      description: Search, for a specific period of time, the number of transactions
+        and the volume.
+      operationId: x-chain-activity-tops
+      parameters:
+      - description: 'Time span, supported values: 1d, 1mo and 1y'
+        in: query
+        name: timespan
+        required: true
+        type: string
+      - description: From date, supported format 2006-01-02T15:04:05Z07:00
+        in: query
+        name: from
+        required: true
+        type: string
+      - description: To date, supported format 2006-01-02T15:04:05Z07:00
+        in: query
+        name: to
+        required: true
+        type: string
+      - description: Search by appId
+        in: query
+        name: appId
+        type: string
+      - description: Search by sourceChain
+        in: query
+        name: sourceChain
+        type: string
+      - description: Search by targetChain
+        in: query
+        name: targetChain
+        type: string
+      responses:
+        "200":
+          description: OK
+          schema:
+            items:
+              $ref: '#/definitions/transactions.ChainActivityTopResult'
+            type: array
+        "400":
+          description: Bad Request
+        "500":
+          description: Internal Server Error
+      tags:
+      - wormholescan
   /swagger.json:
     get:
       description: Returns the swagger specification for this API.

api/handlers/transactions/model.go

@@ -143,6 +143,17 @@ type ChainActivityResult struct {
 	Volume             uint64 `mapstructure:"_value" json:"volume"`
 }
 
+type ChainActivityTopResult struct {
+	Time               time.Time `json:"from" mapstructure:"_time"`
+	To                 string    `json:"to" mapstructure:"to"`
+	ChainSourceID      string    `mapstructure:"emitter_chain" json:"emitter_chain"`
+	ChainDestinationID string    `mapstructure:"destination_chain" json:"destination_chain,omitempty"`
+	Volume             uint64    `mapstructure:"volume" json:"volume"`
+	Txs                uint64    `mapstructure:"count" json:"count"`
+}
+
+type ChainActivityTopResults []ChainActivityTopResult
+
 type ChainActivityTimeSpan string
 
 const (
@@ -202,3 +213,25 @@ type TransactionDto struct {
 	Payload                map[string]interface{} `bson:"payload"`
 	StandardizedProperties map[string]interface{} `bson:"standardizedProperties"`
 }
+
+type ChainActivityTopsQuery struct {
+	SourceChain *sdk.ChainID `json:"source_chain"`
+	TargetChain *sdk.ChainID `json:"target_chain"`
+	AppId       string       `json:"app_id"`
+	From        time.Time    `json:"from"`
+	To          time.Time    `json:"to"`
+	Timespan    Timespan     `json:"timespan"`
+}
+
+type Timespan string
+
+const (
+	Hour  Timespan = "1h"
+	Day   Timespan = "1d"
+	Month Timespan = "1mo"
+	Year  Timespan = "1y"
+)
+
+func (t Timespan) IsValid() bool {
+	return t == Hour || t == Day || t == Month || t == Year
+}

api/handlers/transactions/repository.go

@@ -3,6 +3,7 @@ package transactions
 import (
 	"context"
 	"fmt"
+	"github.com/valyala/fasthttp"
 	"strconv"
 	"strings"
 	"sync"
@@ -1048,3 +1049,338 @@ func (r *Repository) ListTransactionsByAddress(
 
 	return documents, nil
 }
+
+func (r *Repository) FindChainActivityTops(ctx *fasthttp.RequestCtx, q ChainActivityTopsQuery) ([]ChainActivityTopResult, error) {
+	query := r.buildChainActivityQueryTops(q)
+	result, err := r.queryAPI.Query(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	if result.Err() != nil {
+		return nil, result.Err()
+	}
+	var response []ChainActivityTopResult
+	for result.Next() {
+		var row ChainActivityTopResult
+		if err = mapstructure.Decode(result.Record().Values(), &row); err != nil {
+			return nil, err
+		}
+		parsedTime, errTime := time.Parse(time.RFC3339Nano, row.To)
+		if errTime == nil {
+			row.To = parsedTime.Format(time.RFC3339)
+		}
+		response = append(response, row)
+	}
+
+	return response, nil
+}
+
+func (r *Repository) buildChainActivityQueryTops(q ChainActivityTopsQuery) string {
+
+	var start, stop string
+
+	switch q.Timespan {
+	case Hour:
+		start = q.From.Truncate(1 * time.Hour).UTC().Format(time.RFC3339)
+		stop = q.To.Truncate(1 * time.Hour).UTC().Format(time.RFC3339)
+	case Day:
+		start = q.From.Truncate(24 * time.Hour).UTC().Format(time.RFC3339)
+		stop = q.To.Truncate(24 * time.Hour).UTC().Format(time.RFC3339)
+	case Month:
+		start = time.Date(q.From.Year(), q.From.Month(), 1, 0, 0, 0, 0, q.From.Location()).UTC().Format(time.RFC3339)
+		stop = time.Date(q.To.Year(), q.To.Month(), 1, 0, 0, 0, 0, q.To.Location()).UTC().Format(time.RFC3339)
+	default:
+		start = time.Date(q.From.Year(), 1, 1, 0, 0, 0, 0, q.From.Location()).UTC().Format(time.RFC3339)
+		stop = time.Date(q.To.Year(), 1, 1, 0, 0, 0, 0, q.To.Location()).UTC().Format(time.RFC3339)
+	}
+
+	filterTargetChain := ""
+	if q.TargetChain != nil {
+		filterTargetChain = "|> filter(fn: (r) => r.destination_chain == \"" + strconv.Itoa(int(*q.TargetChain)) + "\")"
+	}
+
+	filterSourceChain := ""
+	if q.SourceChain != nil {
+		filterSourceChain = "|> filter(fn: (r) => r.emitter_chain == \"" + strconv.Itoa(int(*q.SourceChain)) + "\")"
+	}
+
+	filterAppId := ""
+	if q.AppId != "" {
+		filterAppId = "|> filter(fn: (r) => r.app_id == \"" + q.AppId + "\")"
+	}
+
+	if q.TargetChain == nil && q.AppId == "" {
+		return r.buildQueryChainActivityTopsByEmitter(q, start, stop, filterSourceChain)
+	}
+
+	var query string
+	switch q.Timespan {
+	case Hour:
+		query = r.buildQueryChainActivityHourly(start, stop, filterSourceChain, filterTargetChain, filterAppId)
+	case Day:
+		query = r.buildQueryChainActivityDaily(start, stop, filterSourceChain, filterTargetChain, filterAppId)
+	case Month:
+		query = r.buildQueryChainActivityMonthly(start, stop, filterSourceChain, filterTargetChain, filterAppId)
+	default:
+		query = r.buildQueryChainActivityYearly(start, stop, filterSourceChain, filterTargetChain, filterAppId)
+	}
+	return query
+}
+
+func (r *Repository) buildQueryChainActivityTopsByEmitter(q ChainActivityTopsQuery, start, stop, filterSourceChain string) string {
+
+	measurement := ""
+	switch q.Timespan {
+	case Hour:
+		measurement = "emitter_chain_activity_1h"
+	default:
+		measurement = "emitter_chain_activity_1d"
+	}
+
+	if q.Timespan == Hour || q.Timespan == Day {
+		query := `
+					import "date"
+
+					from(bucket: "%s")
+		  			|> range(start: %s,stop: %s)
+		  			|> filter(fn: (r) => r._measurement == "%s")
+					%s
+					|> pivot(rowKey:["_time","emitter_chain"], columnKey: ["_field"], valueColumn: "_value")
+					|> sort(columns:["emitter_chain","_time"],desc:false)
+				`
+		return fmt.Sprintf(query, r.bucketInfiniteRetention, start, stop, measurement, filterSourceChain)
+	}
+
+	if q.Timespan == Month {
+		query := `
+				import "date"
+				import "join"
+
+				data = from(bucket: "%s")
+		  				|> range(start: %s,stop: %s)
+		  				|> filter(fn: (r) => r._measurement == "%s")
+						%s
+						|> drop(columns:["to"])
+						|> window(every: 1mo, period:1mo)
+						|> drop(columns:["_time"])
+						|> rename(columns: {_start: "_time"})
+						|> map(fn: (r) => ({r with to: string(v: r._stop)}))
+
+				vols = data		
+						|> filter(fn: (r) => (r._field == "volume" and r._value > 0))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "volume"})
+
+				counts = data
+						|> filter(fn: (r) => (r._field == "count"))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "count"})
+
+				join.inner(
+					    left: vols,
+					    right: counts,
+					    on: (l, r) => l._time == r._time and l.emitter_chain == r.emitter_chain,
+					    as: (l, r) => ({l with count: r.count}),
+				)
+				|> group()
+				|> sort(columns:["emitter_chain","_time"],desc:false)
+				`
+		return fmt.Sprintf(query, r.bucketInfiniteRetention, start, stop, measurement, filterSourceChain)
+	}
+
+	query := `
+				import "date"
+				import "join"
+
+				data = from(bucket: "%s")
+		  				|> range(start: %s,stop: %s)
+		  				|> filter(fn: (r) => r._measurement == "%s")
+						%s
+						|> drop(columns:["to"])
+						|> window(every: 1y, period:1y)
+						|> drop(columns:["_time"])
+						|> rename(columns: {_start: "_time"})
+						|> map(fn: (r) => ({r with to: string(v: r._stop)}))
+
+				vols = data
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "volume"})
+
+				counts = data
+						|> filter(fn: (r) => (r._field == "count"))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "count"})
+
+				join.inner(
+					    left: vols,
+					    right: counts,
+					    on: (l, r) => l._time == r._time and l.emitter_chain == r.emitter_chain,
+					    as: (l, r) => ({l with count: r.count}),
+				)
+				|> group()
+				|> sort(columns:["emitter_chain","_time"],desc:false)
+		`
+	return fmt.Sprintf(query, r.bucketInfiniteRetention, start, stop, measurement, filterSourceChain)
+
+}
+
+func (r *Repository) buildQueryChainActivityHourly(start, stop, filterSourceChain, filterTargetChain, filterAppId string) string {
+	query := `
+					import "date"
+					import "join"
+
+					data = from(bucket: "%s")
+		  			|> range(start: %s,stop: %s)
+		  			|> filter(fn: (r) => r._measurement == "chain_activity_1h")
+					%s
+					%s
+					%s
+					|> drop(columns:["destination_chain"])
+
+					vols = data		
+						|> filter(fn: (r) => (r._field == "volume" and r._value > 0))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "volume"})
+
+					counts = data
+						|> filter(fn: (r) => (r._field == "count"))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "count"})
+
+					join.inner(
+					    left: vols,
+					    right: counts,
+					    on: (l, r) => l._time == r._time and l.to == r.to and l.emitter_chain == r.emitter_chain,
+					    as: (l, r) => ({l with count: r.count}),
+					)
+					|> group()
+					|> sort(columns:["emitter_chain","_time"],desc:false)
+				`
+	return fmt.Sprintf(query, r.bucketInfiniteRetention, start, stop, filterSourceChain, filterTargetChain, filterAppId)
+}
+
+func (r *Repository) buildQueryChainActivityDaily(start, stop, filterSourceChain, filterTargetChain, filterAppId string) string {
+
+	query := `
+					import "date"
+					import "join"
+
+					data = from(bucket: "%s")
+		  			|> range(start: %s,stop: %s)
+		  			|> filter(fn: (r) => r._measurement == "chain_activity_1d")
+					%s
+					%s
+					%s
+					|> drop(columns:["destination_chain"])
+
+					vols = data		
+						|> filter(fn: (r) => (r._field == "volume" and r._value > 0))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "volume"})
+
+					counts = data
+						|> filter(fn: (r) => (r._field == "count"))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "count"})
+
+					join.inner(
+					    left: vols,
+					    right: counts,
+					    on: (l, r) => l._time == r._time and l.to == r.to and l.emitter_chain == r.emitter_chain,
+					    as: (l, r) => ({l with count: r.count}),
+					)
+					|> group()
+					|> sort(columns:["emitter_chain","_time"],desc:false)
+				`
+	return fmt.Sprintf(query, r.bucketInfiniteRetention, start, stop, filterSourceChain, filterTargetChain, filterAppId)
+}
+
+func (r *Repository) buildQueryChainActivityMonthly(start, stop, filterSourceChain, filterTargetChain, filterAppId string) string {
+	query := `
+				import "date"
+				import "join"
+
+				data = from(bucket: "%s")
+		  				|> range(start: %s,stop: %s)
+		  				|> filter(fn: (r) => r._measurement == "chain_activity_1d")
+						%s
+						%s
+						%s
+						|> drop(columns:["destination_chain","to","app_id"])
+						|> window(every: 1mo, period:1mo)
+						|> drop(columns:["_time"])
+						|> rename(columns: {_start: "_time"})
+						|> map(fn: (r) => ({r with to: string(v: r._stop)}))
+
+				vols = data		
+						|> filter(fn: (r) => (r._field == "volume" and r._value > 0))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "volume"})
+
+				counts = data
+						|> filter(fn: (r) => (r._field == "count"))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "count"})
+
+				join.inner(
+					    left: vols,
+					    right: counts,
+					    on: (l, r) => l._time == r._time and l.emitter_chain == r.emitter_chain,
+					    as: (l, r) => ({l with count: r.count}),
+				)
+				|> group()
+				|> sort(columns:["emitter_chain","_time"],desc:false)
+		`
+	return fmt.Sprintf(query, r.bucketInfiniteRetention, start, stop, filterSourceChain, filterTargetChain, filterAppId)
+}
+
+func (r *Repository) buildQueryChainActivityYearly(start, stop, filterSourceChain, filterTargetChain, filterAppId string) string {
+	query := `
+				import "date"
+				import "join"
+
+				data = from(bucket: "%s")
+		  				|> range(start: %s,stop: %s)
+		  				|> filter(fn: (r) => r._measurement == "chain_activity_1d")
+						%s
+						%s
+						%s
+						|> drop(columns:["destination_chain","to","app_id"])
+						|> window(every: 1y, period:1y)
+						|> drop(columns:["_time"])
+						|> rename(columns: {_start: "_time"})
+						|> map(fn: (r) => ({r with to: string(v: r._stop)}))
+
+				vols = data		
+						|> filter(fn: (r) => (r._field == "volume" and r._value > 0))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "volume"})
+
+				counts = data
+						|> filter(fn: (r) => (r._field == "count"))
+						|> group(columns:["_time","to","emitter_chain"])
+						|> sum()
+						|> rename(columns: {_value: "count"})
+
+				join.inner(
+					    left: vols,
+					    right: counts,
+					    on: (l, r) => l._time == r._time and l.emitter_chain == r.emitter_chain,
+					    as: (l, r) => ({l with count: r.count}),
+				)
+				|> group()
+				|> sort(columns:["emitter_chain","_time"],desc:false)
+		`
+	return fmt.Sprintf(query, r.bucketInfiniteRetention, start, stop, filterSourceChain, filterTargetChain, filterAppId)
+}

api/handlers/transactions/service.go

@@ -2,12 +2,13 @@ package transactions
 
 import (
 	"context"
+	errors "errors"
 	"fmt"
+	"github.com/valyala/fasthttp"
 	"strings"
 	"time"
 
 	"github.com/wormhole-foundation/wormhole-explorer/api/cacheable"
-	"github.com/wormhole-foundation/wormhole-explorer/api/internal/errors"
 	errs "github.com/wormhole-foundation/wormhole-explorer/api/internal/errors"
 	"github.com/wormhole-foundation/wormhole-explorer/api/internal/metrics"
 	"github.com/wormhole-foundation/wormhole-explorer/api/internal/pagination"
@@ -34,6 +35,7 @@ const (
 	topAssetsByVolumeKey           = "wormscan:top-assets-by-volume"
 	topChainPairsByNumTransfersKey = "wormscan:top-chain-pairs-by-num-transfers"
 	chainActivityKey               = "wormscan:chain-activity"
+	chainActivityTopsKey           = "wormscan:chain-activity-tops"
 )
 
 // NewService create a new Service.
@@ -157,7 +159,7 @@ func (s *Service) GetTransactionByID(
 		return nil, err
 	}
 	if len(output) == 0 {
-		return nil, errors.ErrNotFound
+		return nil, errs.ErrNotFound
 	}
 
 	// Return matching document
@@ -167,3 +169,44 @@ func (s *Service) GetTransactionByID(
 func (s *Service) GetTokenProvider() *domain.TokenProvider {
 	return s.tokenProvider
 }
+
+func (s *Service) GetChainActivityTops(ctx *fasthttp.RequestCtx, q ChainActivityTopsQuery) (ChainActivityTopResults, error) {
+
+	timeDuration := q.To.Sub(q.From)
+
+	if q.Timespan == Hour && timeDuration > 15*24*time.Hour {
+		return nil, errors.New("time range is too large for hourly data. Max time range allowed: 15 days")
+	}
+
+	if q.Timespan == Day {
+		if timeDuration < 24*time.Hour {
+			return nil, errors.New("time range is too small for daily data. Min time range allowed: 2 day")
+		}
+
+		if timeDuration > 365*24*time.Hour {
+			return nil, errors.New("time range is too large for daily data. Max time range allowed: 1 year")
+		}
+	}
+
+	if q.Timespan == Month {
+		if timeDuration < 30*24*time.Hour {
+			return nil, errors.New("time range is too small for monthly data. Min time range allowed: 60 days")
+		}
+
+		if timeDuration > 10*365*24*time.Hour {
+			return nil, errors.New("time range is too large for monthly data. Max time range allowed: 1 year")
+		}
+	}
+
+	if q.Timespan == Year {
+		if timeDuration < 365*24*time.Hour {
+			return nil, errors.New("time range is too small for yearly data. Min time range allowed: 1 year")
+		}
+
+		if timeDuration > 10*365*24*time.Hour {
+			return nil, errors.New("time range is too large for yearly data. Max time range allowed: 10 year")
+		}
+	}
+
+	return s.repo.FindChainActivityTops(ctx, q)
+}

api/middleware/extract_parameters.go

@@ -61,10 +61,6 @@ func ExtractToChain(c *fiber.Ctx, l *zap.Logger) (*sdk.ChainID, error) {
 	return &result, nil
 }
 
-func ExtractChain(c *fiber.Ctx, l *zap.Logger) (*sdk.ChainID, error) {
-	return extractChainQueryParam(c, l, "chain")
-}
-
 func ExtractSourceChain(c *fiber.Ctx, l *zap.Logger) (*sdk.ChainID, error) {
 	return extractChainQueryParam(c, l, "sourceChain")
 }
@@ -74,12 +70,10 @@ func ExtractTargetChain(c *fiber.Ctx, l *zap.Logger) (*sdk.ChainID, error) {
 }
 
 func extractChainQueryParam(c *fiber.Ctx, l *zap.Logger, queryParam string) (*sdk.ChainID, error) {
-
 	param := c.Query(queryParam)
 	if param == "" {
 		return nil, nil
 	}
-
 	chain, err := strconv.ParseInt(param, 10, 16)
 	if err != nil {
 		requestID := fmt.Sprintf("%v", c.Locals("requestid"))
@@ -90,7 +84,6 @@ func extractChainQueryParam(c *fiber.Ctx, l *zap.Logger, queryParam string) (*sd
 
 		return nil, response.NewInvalidParamError(c, "INVALID CHAIN VALUE", errors.WithStack(err))
 	}
-
 	result := sdk.ChainID(chain)
 	return &result, nil
 }
@@ -358,14 +351,13 @@ func ExtractTimeSpanAndSampleRate(c *fiber.Ctx, l *zap.Logger) (string, string,
 	return timeSpan, sampleRate, nil
 }
 
-func ExtractTime(c *fiber.Ctx, queryParam string) (*time.Time, error) {
+func ExtractTime(c *fiber.Ctx, timeLayout, queryParam string) (*time.Time, error) {
 	// get the start_time from query params
 	date := c.Query(queryParam, "")
 	if date == "" {
 		return nil, nil
 	}
-
-	t, err := time.Parse("20060102T150405Z", date)
+	t, err := time.Parse(timeLayout, date)
 	if err != nil {
 		return nil, response.NewInvalidQueryParamError(c, fmt.Sprintf("INVALID <%s> QUERY PARAMETER", queryParam), nil)
 	}

api/routes/wormscan/routes.go

@@ -85,6 +85,7 @@ func RegisterRoutes(
 	api.Get("/last-txs", transactionCtrl.GetLastTransactions)
 	api.Get("/scorecards", transactionCtrl.GetScorecards)
 	api.Get("/x-chain-activity", transactionCtrl.GetChainActivity)
+	api.Get("/x-chain-activity/tops", transactionCtrl.GetChainActivityTops)
 	api.Get("/top-assets-by-volume", transactionCtrl.GetTopAssets)
 	api.Get("/top-chain-pairs-by-num-transfers", transactionCtrl.GetTopChainPairs)
 	api.Get("token/:chain/:token_address", transactionCtrl.GetTokenByChainAndAddress)

api/routes/wormscan/transactions/controller.go

@@ -2,6 +2,7 @@ package transactions
 
 import (
 	"strconv"
+	"time"
 
 	"github.com/gofiber/fiber/v2"
 	"github.com/shopspring/decimal"
@@ -182,6 +183,75 @@ func (c *Controller) GetTopAssets(ctx *fiber.Ctx) error {
 	return ctx.JSON(response)
 }
 
+// GetChainActivityTops godoc
+// @Description Search for a specific period of time the number of transactions and the volume.
+// @Tags wormholescan
+// @ID x-chain-activity-tops
+// @Method Get
+// @Param timespan query string true "Time span, supported values: 1d, 1mo and 1y"
+// @Param from query string true "From date, supported format 2006-01-02T15:04:05Z07:00"
+// @Param to query string true "To date, supported format 2006-01-02T15:04:05Z07:00"
+// @Param appId query string false "Search by appId"
+// @Param sourceChain query string false "Search by sourceChain"
+// @Param targetChain query string false "Search by targetChain"
+// @Success 200 {object} transactions.ChainActivityTopResults
+// @Failure 400
+// @Failure 500
+// @Router /api/v1/x-chain-activity/tops [get]
+func (c *Controller) GetChainActivityTops(ctx *fiber.Ctx) error {
+
+	sourceChain, err := middleware.ExtractSourceChain(ctx, c.logger)
+	if err != nil {
+		return err
+	}
+	targetChain, err := middleware.ExtractTargetChain(ctx, c.logger)
+	if err != nil {
+		return err
+	}
+	from, err := middleware.ExtractTime(ctx, time.RFC3339, "from")
+	if err != nil {
+		return err
+	}
+	to, err := middleware.ExtractTime(ctx, time.RFC3339, "to")
+	if err != nil {
+		return err
+	}
+	if from == nil || to == nil {
+		return response.NewInvalidParamError(ctx, "missing from/to query params ", nil)
+	}
+
+	payload := transactions.ChainActivityTopsQuery{
+		SourceChain: sourceChain,
+		TargetChain: targetChain,
+		From:        *from,
+		To:          *to,
+		AppId:       middleware.ExtractAppId(ctx, c.logger),
+		Timespan:    transactions.Timespan(ctx.Query("timespan")),
+	}
+
+	if !payload.Timespan.IsValid() {
+		return response.NewInvalidParamError(ctx, "invalid timespan", nil)
+	}
+
+	nowUTC := time.Now().UTC()
+	if nowUTC.Before(payload.To.UTC()) {
+		payload.To = nowUTC
+	}
+
+	if payload.To.Sub(payload.From) <= 0 {
+		return response.NewInvalidParamError(ctx, "invalid time range", nil)
+	}
+
+	// Get the chain activity.
+	activity, err := c.srv.GetChainActivityTops(ctx.Context(), payload)
+	if err != nil {
+		c.logger.Error("Error getting chain activity", zap.Error(err))
+		return err
+	}
+
+	return ctx.JSON(activity)
+}
+
 // GetChainActivity godoc
 // @Description Returns a list of chain pairs by origin chain and destination chain.
 // @Description The list could be rendered by notional or transaction count.

deploy/common/env/production-mainnet.env

@@ -35,3 +35,5 @@ PROTOCOLS_STATS_VERSION=v1
 PROTOCOLS_ACTIVITY_VERSION=v1
 # rpc provider json
 RPC_PROVIDER_JSON=
+GUARDIAN_PROVIDER_JSON=
+

deploy/common/env/production-testnet.env

@@ -39,3 +39,4 @@ PROTOCOLS_STATS_VERSION=v1
 PROTOCOLS_ACTIVITY_VERSION=v1
 # rpc provider json
 RPC_PROVIDER_JSON=
+GUARDIAN_PROVIDER_JSON=

deploy/common/env/staging-mainnet.env

@@ -35,3 +35,4 @@ PROTOCOLS_STATS_VERSION=v1
 PROTOCOLS_ACTIVITY_VERSION=v1
 # rpc provider json
 RPC_PROVIDER_JSON=
+GUARDIAN_PROVIDER_JSON=

deploy/common/env/staging-testnet.env

@@ -39,3 +39,4 @@ PROTOCOLS_STATS_VERSION=v1
 PROTOCOLS_ACTIVITY_VERSION=v1
 # rpc provider json
 RPC_PROVIDER_JSON=
+GUARDIAN_PROVIDER_JSON=

deploy/common/guardian-provider.yaml

@@ -0,0 +1,9 @@
+---
+kind: Secret
+apiVersion: v1
+metadata:
+  name: guardian-provider
+  namespace: {{ .NAMESPACE }}
+type: Opaque
+data:
+  guardian-provider.json: {{ .GUARDIAN_PROVIDER_JSON | b64enc }}

deploy/fly-event-processor/configmap.yaml

@@ -0,0 +1,9 @@
+---
+kind: ConfigMap
+apiVersion: v1
+metadata:
+  name: fly-event-processor
+  namespace: {{ .NAMESPACE }}
+data:
+  aws-region: {{ .SQS_AWS_REGION }}
+  duplicate-vaa-sqs-url: {{ .DUPLICATE_VAA_SQS_URL }}

deploy/fly-event-processor/env/production-mainnet.env

@@ -0,0 +1,17 @@
+ENVIRONMENT=production-mainnet
+NAMESPACE=wormscan
+NAME=wormscan-fly-event-processor
+REPLICAS=2
+IMAGE_NAME=
+RESOURCES_LIMITS_MEMORY=256Mi
+RESOURCES_LIMITS_CPU=500m
+RESOURCES_REQUESTS_MEMORY=128Mi
+RESOURCES_REQUESTS_CPU=250m
+DUPLICATE_VAA_SQS_URL=
+SQS_AWS_REGION=
+P2P_NETWORK=mainnet
+PPROF_ENABLED=false
+AWS_IAM_ROLE=
+ALERT_ENABLED=false
+METRICS_ENABLED=true
+CONSUMER_WORKER_SIZE=1

deploy/fly-event-processor/env/production-testnet.env

@@ -0,0 +1,17 @@
+ENVIRONMENT=production-testnet
+NAMESPACE=wormscan-testnet
+NAME=wormscan-fly-event-processor
+REPLICAS=1
+IMAGE_NAME=
+RESOURCES_LIMITS_MEMORY=30Mi
+RESOURCES_LIMITS_CPU=20m
+RESOURCES_REQUESTS_MEMORY=15Mi
+RESOURCES_REQUESTS_CPU=10m
+DUPLICATE_VAA_SQS_URL=
+SQS_AWS_REGION=
+P2P_NETWORK=testnet
+PPROF_ENABLED=false
+AWS_IAM_ROLE=
+ALERT_ENABLED=false
+METRICS_ENABLED=true
+CONSUMER_WORKER_SIZE=1

deploy/fly-event-processor/env/staging-mainnet.env

@@ -0,0 +1,17 @@
+ENVIRONMENT=staging-mainnet
+NAMESPACE=wormscan
+NAME=wormscan-fly-event-processor
+REPLICAS=2
+IMAGE_NAME=
+RESOURCES_LIMITS_MEMORY=256Mi
+RESOURCES_LIMITS_CPU=500m
+RESOURCES_REQUESTS_MEMORY=128Mi
+RESOURCES_REQUESTS_CPU=250m
+DUPLICATE_VAA_SQS_URL=
+SQS_AWS_REGION=
+P2P_NETWORK=mainnet
+PPROF_ENABLED=true
+AWS_IAM_ROLE=
+ALERT_ENABLED=false
+METRICS_ENABLED=true
+CONSUMER_WORKER_SIZE=1

deploy/fly-event-processor/env/staging-testnet.env

@@ -0,0 +1,17 @@
+ENVIRONMENT=staging-testnet
+NAMESPACE=wormscan-testnet
+NAME=wormscan-fly-event-processor
+REPLICAS=1
+IMAGE_NAME=
+RESOURCES_LIMITS_MEMORY=30Mi
+RESOURCES_LIMITS_CPU=20m
+RESOURCES_REQUESTS_MEMORY=15Mi
+RESOURCES_REQUESTS_CPU=10m
+DUPLICATE_VAA_SQS_URL=
+SQS_AWS_REGION=
+P2P_NETWORK=testnet
+PPROF_ENABLED=false
+AWS_IAM_ROLE=
+ALERT_ENABLED=false
+METRICS_ENABLED=true
+CONSUMER_WORKER_SIZE=1

deploy/fly-event-processor/fly-event-processor-keda.yaml

@@ -0,0 +1,27 @@
+apiVersion: keda.sh/v1alpha1
+kind: TriggerAuthentication
+metadata:
+  name: keda-auth-aws-{{ .NAME }}
+  namespace: {{ .NAMESPACE }}
+spec:
+  podIdentity:
+    provider: aws
+---
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+  name: {{ .NAME }}
+  namespace: {{ .NAMESPACE }}
+spec:
+  scaleTargetRef:
+    name: {{ .NAME }}
+  minReplicaCount: 2
+  maxReplicaCount: 10
+  triggers:
+  - type: aws-sqs-queue
+    authenticationRef:
+      name: keda-auth-aws-{{ .NAME }}
+    metadata:
+      awsRegion: {{ .SQS_AWS_REGION }}
+      queueURL: {{ .DUPLICATE_VAA_SQS_URL }}
+      queueLength: "5"

deploy/fly-event-processor/fly-event-processor-service.yaml

@@ -0,0 +1,102 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .NAME }}
+  namespace: {{ .NAMESPACE }}
+spec:
+  replicas: {{ .REPLICAS }}
+  selector:
+    matchLabels:
+      app: {{ .NAME }}
+  template:
+    metadata:
+      labels:
+        app: {{ .NAME }}
+      annotations:
+        prometheus.io/scrape: "true"
+        prometheus.io/port: "8000"
+    spec:
+      restartPolicy: Always
+      terminationGracePeriodSeconds: 40
+      serviceAccountName: fly-events-processor
+      containers:
+        - name: {{ .NAME }}
+          image: {{ .IMAGE_NAME }}
+          imagePullPolicy: Always
+          volumeMounts:
+            - name: fly-event-processor-config
+              mountPath: /opt/fly-event-processor
+          readinessProbe:
+            initialDelaySeconds: 30
+            periodSeconds: 20
+            timeoutSeconds: 3
+            failureThreshold: 3
+            httpGet:
+              path: /api/ready
+              port: 8000
+          livenessProbe:
+            initialDelaySeconds: 30
+            periodSeconds: 30
+            timeoutSeconds: 3
+            failureThreshold: 3
+            httpGet:
+              path: /api/health
+              port: 8000
+          env:
+            - name: ENVIRONMENT
+              value: {{ .ENVIRONMENT }}
+            - name: PORT
+              value: "8000"
+            - name: LOG_LEVEL
+              value: "INFO"
+            - name: MONGODB_URI
+              valueFrom:
+                secretKeyRef:
+                  name: mongodb
+                  key: mongo-uri
+            - name: MONGODB_DATABASE
+              valueFrom:
+                configMapKeyRef:
+                  name: config
+                  key: mongo-database
+            - name: DUPLICATE_VAA_SQS_URL
+              valueFrom:
+                configMapKeyRef:
+                  name: fly-event-processor
+                  key: duplicate-vaa-sqs-url
+            - name: AWS_REGION
+              valueFrom:
+                configMapKeyRef:
+                  name: fly-event-processor
+                  key: aws-region
+            - name: PPROF_ENABLED
+              value: "{{ .PPROF_ENABLED }}"
+            - name: P2P_NETWORK
+              value: {{ .P2P_NETWORK }}
+            - name: ALERT_ENABLED
+              value: "{{ .ALERT_ENABLED }}"
+            - name: ALERT_API_KEY
+              valueFrom:
+                secretKeyRef:
+                  name: opsgenie
+                  key: api-key
+            - name: METRICS_ENABLED
+              value: "{{ .METRICS_ENABLED }}"
+            - name: CONSUMER_WORKER_SIZE
+              value: "{{ .CONSUMER_WORKER_SIZE }}"
+            - name: GUARDIAN_API_PROVIDER_PATH
+              value: "/opt/fly-event-processor/guardian-provider.json"
+          resources:
+            limits:
+              memory: {{ .RESOURCES_LIMITS_MEMORY }}
+              cpu: {{ .RESOURCES_LIMITS_CPU }}
+            requests:
+              memory: {{ .RESOURCES_REQUESTS_MEMORY }}
+              cpu: {{ .RESOURCES_REQUESTS_CPU }}
+      volumes:
+        - name: fly-event-processor-config
+          secret:
+            secretName: guardian-provider
+            items:
+            - key: guardian-provider.json
+              path: guardian-provider.json

deploy/fly-event-processor/sa.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: fly-events-processor
+  namespace: {{ .NAMESPACE }}
+  annotations:
+    eks.amazonaws.com/role-arn: {{ .AWS_IAM_ROLE }}

fly-event-processor/go.mod

@@ -9,14 +9,20 @@ require (
 	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/prometheus/client_golang v1.16.0
 	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
+	github.com/wormhole-foundation/wormhole/sdk v0.0.0-20240416174455-25e60611a867
 	go.mongodb.org/mongo-driver v1.11.2
 	go.uber.org/zap v1.27.0
+	golang.org/x/net v0.21.0
+	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
 )
 
 require (
+	github.com/algorand/go-algorand-sdk v1.23.0 // indirect
+	github.com/algorand/go-codec/codec v1.1.8 // indirect
 	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
@@ -31,12 +37,13 @@ require (
 	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/certusone/wormhole/node v0.0.0-20240416174455-25e60611a867 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cosmos/btcutil v1.0.5 // 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
@@ -52,9 +59,8 @@ require (
 	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/mr-tron/base58 v1.2.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
@@ -63,14 +69,12 @@ require (
 	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

fly-event-processor/go.sum

@@ -1,3 +1,8 @@
+github.com/algorand/go-algorand-sdk v1.23.0 h1:wlEV6OgDVc/sLeF2y41bwNG/Lr8EoMnN87Ur8N2Gyyo=
+github.com/algorand/go-algorand-sdk v1.23.0/go.mod h1:7i2peZBcE48kfoxNZnLA+mklKh812jBKvQ+t4bn0KBQ=
+github.com/algorand/go-codec v1.1.8/go.mod h1:XhzVs6VVyWMLu6cApb9/192gBjGRVGm5cX5j203Heg4=
+github.com/algorand/go-codec/codec v1.1.8 h1:lsFuhcOH2LiEhpBH3BVUUkdevVmwCRyvb7FCAAPeY6U=
+github.com/algorand/go-codec/codec v1.1.8/go.mod h1:tQ3zAJ6ijTps6V+wp8KsGDnPC2uhHVC7ANyrtkIY0bA=
 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=
@@ -35,11 +40,16 @@ 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 v0.22.1 h1:CnwP9LM/M9xuRrGSCGeMVs9iv09uMqwsVX7EeIpgV2c=
 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/certusone/wormhole/node v0.0.0-20240416174455-25e60611a867 h1:Wdd/ZJuGD3logxkNuT3hA2aq0Uk5uDGMGhca+S1CDnM=
+github.com/certusone/wormhole/node v0.0.0-20240416174455-25e60611a867/go.mod h1:vJHIhQ0MeHZfQ4OpGiUCm3LD3nrdfT1CEIh2JaPCCso=
 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/cosmos/btcutil v1.0.5 h1:t+ZFcX77LpKtDBhjucvnOH8C2l2ioGsBNEQ3jef8xFk=
+github.com/cosmos/btcutil v1.0.5/go.mod h1:IyB7iuqZMJlthe2tkIFL33xPyzbFYP0XVdS8P5lUPis=
 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=
@@ -56,7 +66,6 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cu
 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=
@@ -125,10 +134,11 @@ github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zk
 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/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
+github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 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=
@@ -159,6 +169,7 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 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/test-go/testify v1.1.4 h1:Tf9lntrKUMHiXQ07qBScBTSA0dhYQlu83hswqelv1iE=
 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=
@@ -181,6 +192,7 @@ github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6
 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.1/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 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=
@@ -197,12 +209,14 @@ golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5y
 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.20220106191415-9b9b3d81d5e3/go.mod h1:3p9vT2HGsQu2K1YbXdKPJLVgG5VJdoTa1poYQBtP1AY=
 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-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 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=
@@ -226,10 +240,10 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 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-20211019181941-9d821ace8654/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=
@@ -251,19 +265,26 @@ 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.10/go.mod h1:Uh6Zz+xoGYZom868N8YTex3t7RhtHDBrE8Gzo9bV56E=
 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-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/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/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 h1:VpOs+IwYnYBaFnrNAeB8UUWtL3vEUnzSCL1nVjPhqrw=
+gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA=
 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.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 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/processor/processor.go

@@ -119,7 +119,7 @@ func (p *Processor) Process(ctx context.Context, params *Params) error {
 	}
 
 	// 2.2 Get all duplicate vaas by vaaId
-	duplicateVaaDocs, err := p.repository.FindDuplicateVAAs(ctx, params.TrackID)
+	duplicateVaaDocs, err := p.repository.FindDuplicateVAAs(ctx, params.VaaID)
 	if err != nil {
 		logger.Error("error getting duplicate vaas from collection", zap.Error(err))
 		return err

fly/cmd/backfiller/txhash_encoding_strategy.go

@@ -9,6 +9,7 @@ import (
 	"github.com/wormhole-foundation/wormhole-explorer/common/dbutil"
 	"github.com/wormhole-foundation/wormhole-explorer/common/domain"
 	"github.com/wormhole-foundation/wormhole-explorer/common/logger"
+	"github.com/wormhole-foundation/wormhole-explorer/fly/event"
 	"github.com/wormhole-foundation/wormhole-explorer/fly/internal/metrics"
 	"github.com/wormhole-foundation/wormhole-explorer/fly/producer"
 	"github.com/wormhole-foundation/wormhole-explorer/fly/storage"
@@ -41,6 +42,7 @@ func RunTxHashEncoding(cfg TxHashEncondingConfig) {
 		db.Database,
 		producer.NewVAAInMemory(logger).Push,
 		txhash.NewMongoTxHash(db.Database, logger),
+		event.NewNoopEventDispatcher(),
 		logger)
 
 	workerTxHashEncoding(ctx, logger, repository, vaa.ChainID(cfg.ChainID), cfg.PageSize)