Add backfiller command to process vaa saved in database (#349)
* Add endpoint to parse a vaa * Add backfiller command to process vaa saved in database
This commit is contained in:
ftocal
GitHub
parent
69251f136d
commit
5c2f91d935
|
|
@ -37,7 +37,6 @@ require (
|
|||
github.com/golang/snappy v0.0.4 // indirect
|
||||
github.com/google/uuid v1.3.0 // indirect
|
||||
github.com/holiman/uint256 v1.2.1 // indirect
|
||||
github.com/influxdata/influxdb-client-go v1.4.0
|
||||
github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097 // indirect
|
||||
github.com/klauspost/compress v1.15.9 // indirect
|
||||
github.com/mattn/go-colorable v0.1.13 // indirect
|
||||
|
|
|
|||
|
|
@ -42,7 +42,6 @@ github.com/decred/dcrd/crypto/blake256 v1.0.0 h1:/8DMNYp9SGi5f0w7uCm6d6M4OU2rGFK
|
|||
github.com/decred/dcrd/crypto/blake256 v1.0.0/go.mod h1:sQl2p6Y26YV+ZOcSTP6thNdn47hh8kt6rqSlvmrXFAc=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 h1:YLtO71vCjJRCBcrPMtQ9nqBsqpA1m5sE92cU+pd5Mcc=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1/go.mod h1:hyedUtir6IdtD/7lIxGeCxkaw7y45JueMRL4DIyJDKs=
|
||||
github.com/deepmap/oapi-codegen v1.3.6/go.mod h1:aBozjEveG+33xPiP55Iw/XbVkhtZHEGLq3nxlX0+hfU=
|
||||
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=
|
||||
|
|
@ -51,10 +50,8 @@ 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.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
|
||||
github.com/getkin/kin-openapi v0.2.0/go.mod h1:V1z9xl9oF5Wt7v32ne4FmiF1alpS4dM6mNzoywPOXlk=
|
||||
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 v4.0.2+incompatible/go.mod h1:eB3wogJHnLi3x/kFX2A+IbTBlXxmMeXJVKy9tTv1XzQ=
|
||||
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=
|
||||
|
|
@ -76,11 +73,8 @@ github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
|
|||
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/influxdata/influxdb-client-go v1.4.0 h1:+KavOkwhLClHFfYcJMHHnTL5CZQhXJzOm5IKHI9BqJk=
|
||||
github.com/influxdata/influxdb-client-go v1.4.0/go.mod h1:S+oZsPivqbcP1S9ur+T+QqXvrYS3NCZeMQtBoH4D1dw=
|
||||
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-20200327222509-2487e7298839/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
|
||||
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=
|
||||
|
|
@ -95,21 +89,18 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN
|
|||
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.1.11/go.mod h1:i541M3Fj6f76NZtHSj7TXnyM8n2gaodfvfxNnFqi74g=
|
||||
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.4/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.10/go.mod h1:qgIWMr58cqv1PHHyhnkY9lrL7etaEgOFcMEpPG5Rm84=
|
||||
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.17 h1:BTarxUcIeDqL27Mc+vyvdWYSL28zpIhv3RoTdsLMPng=
|
||||
|
|
@ -151,7 +142,6 @@ github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyC
|
|||
github.com/valyala/fasthttp v1.44.0 h1:R+gLUhldIsfg1HokMuQjdQ5bh9nuXHPIfvkYUu9eR5Q=
|
||||
github.com/valyala/fasthttp v1.44.0/go.mod h1:f6VbjjoI3z1NDOZOv17o6RvtRSWxC77seBFc2uWtgiY=
|
||||
github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
|
||||
github.com/valyala/fasttemplate v1.1.0/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=
|
||||
|
|
@ -176,9 +166,7 @@ go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9i
|
|||
go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
|
||||
go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191112222119-e1110fd1c708/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
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=
|
||||
|
|
@ -188,7 +176,6 @@ golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0
|
|||
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
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-20191112182307-2180aed22343/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-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
|
|
@ -203,9 +190,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
|
|||
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-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191115151921-52ab43148777/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=
|
||||
|
|
@ -222,7 +207,6 @@ golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXR
|
|||
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.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
|
||||
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=
|
||||
|
|
@ -242,7 +226,6 @@ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8
|
|||
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.5/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=
|
||||
|
|
|
|||
|
|
@ -0,0 +1,42 @@
|
|||
---
|
||||
apiVersion: batch/v1
|
||||
kind: Job
|
||||
metadata:
|
||||
name: {{ .NAME }}-backfiller
|
||||
namespace: {{ .NAMESPACE }}
|
||||
spec:
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app: {{ .NAME }}-backfiller
|
||||
spec:
|
||||
restartPolicy: Never
|
||||
terminationGracePeriodSeconds: 40
|
||||
containers:
|
||||
- name: {{ .NAME }}-backfiller
|
||||
image: {{ .IMAGE_NAME }}
|
||||
imagePullPolicy: Always
|
||||
env:
|
||||
- name: MONGODB_URI
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: mongodb
|
||||
key: mongo-uri
|
||||
- name: MONGODB_DATABASE
|
||||
valueFrom:
|
||||
configMapKeyRef:
|
||||
name: config
|
||||
key: mongo-database
|
||||
command: ["/parser-pipeline"]
|
||||
args:
|
||||
- backfiller
|
||||
- --mongo-uri
|
||||
- "$(MONGODB_URI)"
|
||||
- --mongo-database
|
||||
- "$(MONGODB_DATABASE)"
|
||||
- --vaa-payload-parser-url
|
||||
- "{{ .VAA_PAYLOAD_PARSER_URL }}"
|
||||
- --vaa-payload-parser-timeout
|
||||
- "{{ .VAA_PAYLOAD_PARSER_TIMEOUT }}"
|
||||
- --page-size
|
||||
- "50"
|
||||
|
|
@ -1,19 +1,60 @@
|
|||
# Parser
|
||||
|
||||
## Config sns topic
|
||||
This component is in charge of parsing the VAA payload and persists it.
|
||||
|
||||
VAA parsing is delegated to external service so that users can add custom parsers in the external service without affecting this service.
|
||||
|
||||
## Usage
|
||||
|
||||
### Service
|
||||
|
||||
```bash
|
||||
parser service
|
||||
```
|
||||
|
||||
### Backfiller
|
||||
```bash
|
||||
parser backfiller [flags]
|
||||
```
|
||||
|
||||
#### Command-line arguments
|
||||
- **--end-time** *string* maximum VAA timestamp to process (default now)
|
||||
- **--log-level** *string* log level (default "INFO")
|
||||
- **--mongo-database** *string* mongo database
|
||||
- **--mongo-uri** *string* mongo connection
|
||||
- **--page-size** *int* VAA payload parser timeout (default 100)
|
||||
- **--start-time** *string* minimum VAA timestamp to process (default "1970-01-01T00:00:00Z")
|
||||
- **--vaa-payload-parser-timeout** *int* maximum waiting time in call to VAA payload service in second (default 10)
|
||||
- **--vaa-payload-parser-url** *string* VAA payload parser service URL
|
||||
|
||||
|
||||
## Running parser as service with localstack
|
||||
|
||||
Here are some aws commands to configure localstack with the necessary resources
|
||||
|
||||
### Config sns topic
|
||||
|
||||
```bash
|
||||
aws --profile localstack --endpoint-url=http://localhost:4566 sns create-topic --name vaas-pipeline.fifo --attributes FifoTopic=true,ContentBasedDeduplication=false
|
||||
```
|
||||
### Config SQS FIFO with dead letter queue localstack
|
||||
|
||||
## Config SQS FIFO with dead letter queue localstack
|
||||
|
||||
```bash
|
||||
aws --profile localstack --endpoint-url=http://localhost:4566 sqs create-queue --queue-name=wormhole-vaa-parser-dlq-queue.fifo --attributes "FifoQueue=true"
|
||||
```
|
||||
|
||||
```bash
|
||||
aws --profile localstack --endpoint-url=http://localhost:4566 sqs create-queue --queue-name=wormhole-vaa-parser-queue.fifo --attributes FifoQueue=true,MessageRetentionPeriod=3600,ReceiveMessageWaitTimeSeconds=5,VisibilityTimeout=20,RedrivePolicy="\"{\\\"deadLetterTargetArn\\\":\\\"arn:aws:sqs:us-east-1:000000000000:wormhole-vaa-parser-dlq-queue.fifo\\\",\\\"maxReceiveCount\\\":\\\"2\\\"}\""
|
||||
```
|
||||
|
||||
## Subscribe SQS FIFO to vaas-pipeline.fifo topic
|
||||
### Subscribe SQS FIFO to vaas-pipeline.fifo topic
|
||||
|
||||
```bash
|
||||
aws --profile localstack --endpoint-url=http://localhost:4566 sns subscribe --topic-arn arn:aws:sns:us-east-1:000000000000:vaas-pipeline.fifo --protocol sqs --notification-endpoint http://localhost:4566/000000000000/wormhole-vaa-parser-queue.fifo
|
||||
```
|
||||
|
||||
## Check message in the dead letter queue localstack
|
||||
### Check message in the dead letter queue localstack
|
||||
|
||||
aws --profile localstack --endpoint-url=http://localhost:4566 sqs receive-message --queue-url=http://localhost:4566/000000000000/wormhole-vaa-parser-dlq-queue.fifo
|
||||
```bash
|
||||
aws --profile localstack --endpoint-url=http://localhost:4566 sqs receive-message --queue-url=http://localhost:4566/000000000000/wormhole-vaa-parser-dlq-queue.fifo
|
||||
```
|
||||
|
|
|
|||
|
|
@ -0,0 +1,91 @@
|
|||
package backfiller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/wormhole-foundation/wormhole-explorer/common/logger"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/config"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/http/vaa"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/internal/db"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/parser"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/processor"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
func Run(config *config.BackfillerConfiguration) {
|
||||
|
||||
rootCtx := context.Background()
|
||||
|
||||
logger := logger.New("wormhole-explorer-parser", logger.WithLevel(config.LogLevel))
|
||||
|
||||
logger.Info("Starting wormhole-explorer-parser as backfiller ...")
|
||||
|
||||
startTime, err := time.Parse(time.RFC3339, config.StartTime)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to parse start time", zap.Error(err))
|
||||
}
|
||||
|
||||
endTime := time.Now()
|
||||
if config.EndTime != "" {
|
||||
endTime, err = time.Parse(time.RFC3339, config.EndTime)
|
||||
if err != nil {
|
||||
logger.Fatal("Failed to parse end time", zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
if startTime.After(endTime) {
|
||||
logger.Fatal("Start time should be before end time",
|
||||
zap.String("start_time", startTime.Format(time.RFC3339)), zap.String("end_time", endTime.Format(time.RFC3339)))
|
||||
}
|
||||
|
||||
//setup DB connection
|
||||
db, err := db.New(rootCtx, logger, config.MongoURI, config.MongoDatabase)
|
||||
if err != nil {
|
||||
logger.Fatal("Failed to connect MongoDB", zap.Error(err))
|
||||
}
|
||||
|
||||
parserVAAAPIClient, err := parser.NewParserVAAAPIClient(config.VaaPayloadParserTimeout, config.VaaPayloadParserURL, logger)
|
||||
if err != nil {
|
||||
logger.Fatal("Failed to create parse vaa api client")
|
||||
}
|
||||
|
||||
parserRepository := parser.NewRepository(db.Database, logger)
|
||||
vaaRepository := vaa.NewRepository(db.Database, logger)
|
||||
|
||||
//create a processor
|
||||
processor := processor.New(parserVAAAPIClient, parserRepository, logger)
|
||||
|
||||
logger.Info("Started wormhole-explorer-parser as backfiller")
|
||||
|
||||
//start backfilling
|
||||
page := int64(0)
|
||||
for {
|
||||
logger.Info("Processing page", zap.Int64("page", page),
|
||||
zap.String("start_time", startTime.Format(time.RFC3339)),
|
||||
zap.String("end_time", endTime.Format(time.RFC3339)))
|
||||
|
||||
vaas, err := vaaRepository.FindPageByTimeRange(rootCtx, startTime, endTime, page, config.PageSize)
|
||||
if err != nil {
|
||||
logger.Error("Failed to get vaas", zap.Error(err))
|
||||
break
|
||||
}
|
||||
|
||||
if len(vaas) == 0 {
|
||||
logger.Info("Empty page", zap.Int64("page", page))
|
||||
break
|
||||
}
|
||||
for _, v := range vaas {
|
||||
logger.Debug("Processing vaa", zap.String("id", v.ID))
|
||||
_, err := processor.Process(rootCtx, v.Vaa)
|
||||
if err != nil {
|
||||
logger.Error("Failed to process vaa", zap.String("id", v.ID), zap.Error(err))
|
||||
}
|
||||
}
|
||||
page++
|
||||
}
|
||||
logger.Info("Closing database connections ...")
|
||||
db.Close()
|
||||
|
||||
logger.Info("Finish wormhole-explorer-parser as backfiller")
|
||||
}
|
||||
|
|
@ -1,157 +1,77 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"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"
|
||||
influxdb2 "github.com/influxdata/influxdb-client-go/v2"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/common/logger"
|
||||
"github.com/spf13/cobra"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/cmd/backfiller"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/cmd/service"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/config"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/consumer"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/http/infrastructure"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/internal/db"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/internal/sqs"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/parser"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/processor"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/queue"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type exitCode int
|
||||
|
||||
func handleExit() {
|
||||
if r := recover(); r != nil {
|
||||
if e, ok := r.(exitCode); ok {
|
||||
os.Exit(int(e))
|
||||
}
|
||||
panic(r) // not an Exit, bubble up
|
||||
}
|
||||
}
|
||||
|
||||
func main() {
|
||||
|
||||
defer handleExit()
|
||||
rootCtx, rootCtxCancel := context.WithCancel(context.Background())
|
||||
|
||||
config, err := config.New(rootCtx)
|
||||
if err != nil {
|
||||
log.Fatal("Error creating config", err)
|
||||
}
|
||||
|
||||
logger := logger.New("wormhole-explorer-parser", logger.WithLevel(config.LogLevel))
|
||||
|
||||
logger.Info("Starting wormhole-explorer-parser ...")
|
||||
|
||||
//setup DB connection
|
||||
db, err := db.New(rootCtx, logger, config.MongoURI, config.MongoDatabase)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to connect MongoDB", zap.Error(err))
|
||||
}
|
||||
|
||||
parserVAAAPIClient, err := parser.NewParserVAAAPIClient(config.VaaPayloadParserTimeout,
|
||||
config.VaaPayloadParserURL, logger)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to create parse vaa api client")
|
||||
}
|
||||
|
||||
// get consumer function.
|
||||
sqsConsumer, vaaConsumeFunc := newVAAConsume(rootCtx, config, logger)
|
||||
repository := parser.NewRepository(db.Database, logger)
|
||||
|
||||
//create a processor
|
||||
processor := processor.New(repository, logger)
|
||||
|
||||
// create and start a consumer
|
||||
consumer := consumer.New(vaaConsumeFunc, processor.Process, parserVAAAPIClient, logger)
|
||||
consumer.Start(rootCtx)
|
||||
|
||||
server := infrastructure.NewServer(logger, config.Port, config.PprofEnabled, config.IsQueueConsumer(), sqsConsumer, db.Database)
|
||||
server.Start()
|
||||
|
||||
logger.Info("Started wormhole-explorer-parser")
|
||||
|
||||
// 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()))
|
||||
}
|
||||
|
||||
logger.Info("root context cancelled, exiting...")
|
||||
rootCtxCancel()
|
||||
|
||||
logger.Info("Closing database connections ...")
|
||||
db.Close()
|
||||
logger.Info("Closing Http server ...")
|
||||
server.Stop()
|
||||
logger.Info("Finished wormhole-explorer-parser")
|
||||
execute()
|
||||
}
|
||||
|
||||
func newAwsConfig(appCtx context.Context, cfg *config.Configuration) (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
|
||||
func execute() error {
|
||||
root := &cobra.Command{
|
||||
Use: "parser",
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
if len(args) == 0 {
|
||||
service.Run()
|
||||
}
|
||||
|
||||
return aws.Endpoint{}, &aws.EndpointNotFoundError{}
|
||||
})
|
||||
|
||||
awsCfg, err := awsconfig.LoadDefaultConfig(appCtx,
|
||||
awsconfig.WithRegion(region),
|
||||
awsconfig.WithEndpointResolver(customResolver),
|
||||
awsconfig.WithCredentialsProvider(credentials),
|
||||
)
|
||||
return awsCfg, err
|
||||
}
|
||||
return awsconfig.LoadDefaultConfig(appCtx, awsconfig.WithRegion(region))
|
||||
}
|
||||
|
||||
// Creates a callbacks depending on whether the execution is local (memory queue) or not (SQS queue)
|
||||
func newVAAConsume(appCtx context.Context, config *config.Configuration, logger *zap.Logger) (*sqs.Consumer, queue.VAAConsumeFunc) {
|
||||
sqsConsumer, err := newSQSConsumer(appCtx, config)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to create sqs consumer", zap.Error(err))
|
||||
},
|
||||
}
|
||||
|
||||
filterConsumeFunc := newFilterFunc(config)
|
||||
vaaQueue := queue.NewVAASQS(sqsConsumer, filterConsumeFunc, logger)
|
||||
return sqsConsumer, vaaQueue.Consume
|
||||
addServiceCommand(root)
|
||||
addBackfiller(root)
|
||||
|
||||
return root.Execute()
|
||||
}
|
||||
|
||||
func newSQSConsumer(appCtx context.Context, config *config.Configuration) (*sqs.Consumer, error) {
|
||||
awsconfig, err := newAwsConfig(appCtx, config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
func addServiceCommand(root *cobra.Command) {
|
||||
serviceCommand := &cobra.Command{
|
||||
Use: "service",
|
||||
Short: "Run parser as service",
|
||||
Run: func(_ *cobra.Command, _ []string) {
|
||||
service.Run()
|
||||
},
|
||||
}
|
||||
|
||||
return sqs.NewConsumer(awsconfig, config.SQSUrl,
|
||||
sqs.WithMaxMessages(10),
|
||||
sqs.WithVisibilityTimeout(120))
|
||||
root.AddCommand(serviceCommand)
|
||||
}
|
||||
|
||||
func newFilterFunc(cfg *config.Configuration) queue.FilterConsumeFunc {
|
||||
if cfg.P2pNetwork == config.P2pMainNet {
|
||||
return queue.PythFilter
|
||||
func addBackfiller(root *cobra.Command) {
|
||||
var mongoUri, mongoDb, vaaPayloadParserURL, logLevel, startTime, endTime string
|
||||
var vaaPayloadParserTimeout, pageSize int64
|
||||
|
||||
backfillerCommand := &cobra.Command{
|
||||
Use: "backfiller",
|
||||
Short: "Run backfiller to backfill data",
|
||||
Run: func(_ *cobra.Command, _ []string) {
|
||||
cfg := &config.BackfillerConfiguration{
|
||||
LogLevel: logLevel,
|
||||
MongoURI: mongoUri,
|
||||
MongoDatabase: mongoDb,
|
||||
VaaPayloadParserURL: vaaPayloadParserURL,
|
||||
VaaPayloadParserTimeout: vaaPayloadParserTimeout,
|
||||
StartTime: startTime,
|
||||
EndTime: endTime,
|
||||
PageSize: pageSize,
|
||||
}
|
||||
backfiller.Run(cfg)
|
||||
},
|
||||
}
|
||||
return queue.NonFilter
|
||||
}
|
||||
backfillerCommand.Flags().StringVar(&logLevel, "log-level", "INFO", "log level")
|
||||
backfillerCommand.Flags().StringVar(&mongoUri, "mongo-uri", "", "Mongo connection")
|
||||
backfillerCommand.Flags().StringVar(&mongoDb, "mongo-database", "", "Mongo database")
|
||||
backfillerCommand.Flags().StringVar(&vaaPayloadParserURL, "vaa-payload-parser-url", "", "VAA payload parser service URL")
|
||||
backfillerCommand.Flags().Int64Var(&vaaPayloadParserTimeout, "vaa-payload-parser-timeout", 10, "maximum waiting time in call to VAA payload service in seconds")
|
||||
backfillerCommand.Flags().StringVar(&startTime, "start-time", "1970-01-01T00:00:00Z", "minimum VAA timestamp to process")
|
||||
backfillerCommand.Flags().StringVar(&endTime, "end-time", "", "maximum VAA timestamp to process (default now)")
|
||||
backfillerCommand.Flags().Int64Var(&pageSize, "page-size", 100, "number of documents retrieved at a time")
|
||||
|
||||
func newInfluxClient(url, token string) influxdb2.Client {
|
||||
return influxdb2.NewClient(url, token)
|
||||
backfillerCommand.MarkFlagRequired("mongo-uri")
|
||||
backfillerCommand.MarkFlagRequired("mongo-database")
|
||||
backfillerCommand.MarkFlagRequired("p2p-network")
|
||||
backfillerCommand.MarkFlagRequired("vaa-payload-parser-url")
|
||||
|
||||
root.AddCommand(backfillerCommand)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,155 @@
|
|||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"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/logger"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/config"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/consumer"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/http/infrastructure"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/http/vaa"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/internal/db"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/internal/sqs"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/parser"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/processor"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/queue"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type exitCode int
|
||||
|
||||
func handleExit() {
|
||||
if r := recover(); r != nil {
|
||||
if e, ok := r.(exitCode); ok {
|
||||
os.Exit(int(e))
|
||||
}
|
||||
panic(r) // not an Exit, bubble up
|
||||
}
|
||||
}
|
||||
|
||||
func Run() {
|
||||
|
||||
defer handleExit()
|
||||
rootCtx, rootCtxCancel := context.WithCancel(context.Background())
|
||||
|
||||
config, err := config.New(rootCtx)
|
||||
if err != nil {
|
||||
log.Fatal("Error creating config", err)
|
||||
}
|
||||
|
||||
logger := logger.New("wormhole-explorer-parser", logger.WithLevel(config.LogLevel))
|
||||
|
||||
logger.Info("Starting wormhole-explorer-parser ...")
|
||||
|
||||
//setup DB connection
|
||||
db, err := db.New(rootCtx, logger, config.MongoURI, config.MongoDatabase)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to connect MongoDB", zap.Error(err))
|
||||
}
|
||||
|
||||
parserVAAAPIClient, err := parser.NewParserVAAAPIClient(config.VaaPayloadParserTimeout,
|
||||
config.VaaPayloadParserURL, logger)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to create parse vaa api client")
|
||||
}
|
||||
|
||||
// get consumer function.
|
||||
sqsConsumer, vaaConsumeFunc := newVAAConsume(rootCtx, config, logger)
|
||||
repository := parser.NewRepository(db.Database, logger)
|
||||
|
||||
//create a processor
|
||||
processor := processor.New(parserVAAAPIClient, repository, logger)
|
||||
|
||||
// create and start a consumer
|
||||
consumer := consumer.New(vaaConsumeFunc, processor.Process, logger)
|
||||
consumer.Start(rootCtx)
|
||||
|
||||
vaaRepository := vaa.NewRepository(db.Database, logger)
|
||||
vaaController := vaa.NewController(vaaRepository, processor.Process, logger)
|
||||
server := infrastructure.NewServer(logger, config.Port, config.PprofEnabled, config.IsQueueConsumer(), sqsConsumer, db.Database, vaaController)
|
||||
server.Start()
|
||||
|
||||
logger.Info("Started wormhole-explorer-parser")
|
||||
|
||||
// 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()))
|
||||
}
|
||||
|
||||
logger.Info("root context cancelled, exiting...")
|
||||
rootCtxCancel()
|
||||
|
||||
logger.Info("Closing database connections ...")
|
||||
db.Close()
|
||||
logger.Info("Closing Http server ...")
|
||||
server.Stop()
|
||||
logger.Info("Finished wormhole-explorer-parser")
|
||||
}
|
||||
|
||||
func newAwsConfig(appCtx 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(appCtx,
|
||||
awsconfig.WithRegion(region),
|
||||
awsconfig.WithEndpointResolver(customResolver),
|
||||
awsconfig.WithCredentialsProvider(credentials),
|
||||
)
|
||||
return awsCfg, err
|
||||
}
|
||||
return awsconfig.LoadDefaultConfig(appCtx, awsconfig.WithRegion(region))
|
||||
}
|
||||
|
||||
// Creates a callbacks depending on whether the execution is local (memory queue) or not (SQS queue)
|
||||
func newVAAConsume(appCtx context.Context, config *config.ServiceConfiguration, logger *zap.Logger) (*sqs.Consumer, queue.VAAConsumeFunc) {
|
||||
sqsConsumer, err := newSQSConsumer(appCtx, config)
|
||||
if err != nil {
|
||||
logger.Fatal("failed to create sqs consumer", zap.Error(err))
|
||||
}
|
||||
|
||||
filterConsumeFunc := newFilterFunc(config)
|
||||
vaaQueue := queue.NewVAASQS(sqsConsumer, filterConsumeFunc, logger)
|
||||
return sqsConsumer, vaaQueue.Consume
|
||||
}
|
||||
|
||||
func newSQSConsumer(appCtx context.Context, config *config.ServiceConfiguration) (*sqs.Consumer, error) {
|
||||
awsconfig, err := newAwsConfig(appCtx, config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return sqs.NewConsumer(awsconfig, config.SQSUrl,
|
||||
sqs.WithMaxMessages(10),
|
||||
sqs.WithVisibilityTimeout(120))
|
||||
}
|
||||
|
||||
func newFilterFunc(cfg *config.ServiceConfiguration) queue.FilterConsumeFunc {
|
||||
if cfg.P2pNetwork == config.P2pMainNet {
|
||||
return queue.PythFilter
|
||||
}
|
||||
return queue.NonFilter
|
||||
}
|
||||
|
|
@ -14,8 +14,8 @@ const (
|
|||
P2pDevNet = "devnet"
|
||||
)
|
||||
|
||||
// Configuration represents the application configuration with the default values.
|
||||
type Configuration struct {
|
||||
// ServiceConfiguration represents the application configuration when running as service with default values.
|
||||
type ServiceConfiguration struct {
|
||||
Env string `env:"ENV,default=development"`
|
||||
LogLevel string `env:"LOG_LEVEL,default=INFO"`
|
||||
Port string `env:"PORT,default=8000"`
|
||||
|
|
@ -33,11 +33,23 @@ type Configuration struct {
|
|||
P2pNetwork string `env:"P2P_NETWORK,required"`
|
||||
}
|
||||
|
||||
// BackfillerConfiguration represents the application configuration when running as backfiller with default values.
|
||||
type BackfillerConfiguration struct {
|
||||
LogLevel string `env:"LOG_LEVEL,default=INFO"`
|
||||
MongoURI string `env:"MONGODB_URI,required"`
|
||||
MongoDatabase string `env:"MONGODB_DATABASE,required"`
|
||||
VaaPayloadParserURL string `env:"VAA_PAYLOAD_PARSER_URL, required"`
|
||||
VaaPayloadParserTimeout int64 `env:"VAA_PAYLOAD_PARSER_TIMEOUT, required"`
|
||||
StartTime string `env:"START_TIME"`
|
||||
EndTime string `env:"END_TIME"`
|
||||
PageSize int64 `env:"PAGE_SIZE,default=100"`
|
||||
}
|
||||
|
||||
// New creates a configuration with the values from .env file and environment variables.
|
||||
func New(ctx context.Context) (*Configuration, error) {
|
||||
func New(ctx context.Context) (*ServiceConfiguration, error) {
|
||||
_ = godotenv.Load(".env", "../.env")
|
||||
|
||||
var configuration Configuration
|
||||
var configuration ServiceConfiguration
|
||||
if err := envconfig.Process(ctx, &configuration); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -46,6 +58,6 @@ func New(ctx context.Context) (*Configuration, error) {
|
|||
}
|
||||
|
||||
// IsQueueConsumer check if consumer mode is QUEUE.
|
||||
func (c *Configuration) IsQueueConsumer() bool {
|
||||
func (c *ServiceConfiguration) IsQueueConsumer() bool {
|
||||
return c.ConsumerMode == "QUEUE"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,13 +2,9 @@ package consumer
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/parser"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/processor"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/queue"
|
||||
"github.com/wormhole-foundation/wormhole/sdk/vaa"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
|
|
@ -16,13 +12,12 @@ import (
|
|||
type Consumer struct {
|
||||
consume queue.VAAConsumeFunc
|
||||
process processor.ProcessorFunc
|
||||
parser parser.ParserVAAAPIClient
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// New creates a new vaa consumer.
|
||||
func New(consume queue.VAAConsumeFunc, process processor.ProcessorFunc, parser parser.ParserVAAAPIClient, logger *zap.Logger) *Consumer {
|
||||
return &Consumer{consume: consume, process: process, parser: parser, logger: logger}
|
||||
func New(consume queue.VAAConsumeFunc, process processor.ProcessorFunc, logger *zap.Logger) *Consumer {
|
||||
return &Consumer{consume: consume, process: process, logger: logger}
|
||||
}
|
||||
|
||||
// Start consumes messages from VAA queue, parse and store those messages in a repository.
|
||||
|
|
@ -38,44 +33,7 @@ func (c *Consumer) Start(ctx context.Context) {
|
|||
continue
|
||||
}
|
||||
|
||||
// unmarshal vaa.
|
||||
vaa, err := vaa.Unmarshal(event.Vaa)
|
||||
if err != nil {
|
||||
c.logger.Error("Invalid vaa", zap.String("id", event.ID), zap.Error(err))
|
||||
msg.Failed()
|
||||
continue
|
||||
}
|
||||
|
||||
// call vaa-payload-parser api to parse a VAA.
|
||||
vaaParseResponse, err := c.parser.Parse(event.ChainID, event.EmitterAddress, event.Sequence, vaa.Payload)
|
||||
if err != nil {
|
||||
if errors.Is(err, parser.ErrInternalError) {
|
||||
c.logger.Info("error parsing VAA, will retry later", zap.Uint16("chainID", event.ChainID),
|
||||
zap.String("address", event.EmitterAddress), zap.String("sequence", event.Sequence), zap.Error(err))
|
||||
msg.Failed()
|
||||
continue
|
||||
}
|
||||
|
||||
c.logger.Info("VAA cannot be parsed", zap.Uint16("chainID", event.ChainID),
|
||||
zap.String("address", event.EmitterAddress), zap.String("sequence", event.Sequence), zap.Error(err))
|
||||
msg.Done()
|
||||
continue
|
||||
}
|
||||
|
||||
// create ParsedVaaUpdate to upsert.
|
||||
now := time.Now()
|
||||
vaaParsed := parser.ParsedVaaUpdate{
|
||||
ID: event.ID,
|
||||
EmitterChain: event.ChainID,
|
||||
EmitterAddr: event.EmitterAddress,
|
||||
Sequence: event.Sequence,
|
||||
AppID: vaaParseResponse.AppID,
|
||||
Result: vaaParseResponse.Result,
|
||||
Timestamp: vaa.Timestamp,
|
||||
UpdatedAt: &now,
|
||||
}
|
||||
|
||||
err = c.process(ctx, &vaaParsed)
|
||||
_, err := c.process(ctx, event.Vaa)
|
||||
if err != nil {
|
||||
c.logger.Error("Error processing parsed vaa",
|
||||
zap.String("id", event.ID),
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ require (
|
|||
github.com/aws/aws-sdk-go-v2/config v1.1.1
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.1.1
|
||||
github.com/aws/aws-sdk-go-v2/service/sqs v1.20.2
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.12.2
|
||||
github.com/spf13/cobra v1.7.0
|
||||
github.com/wormhole-foundation/wormhole-explorer/common v0.0.0-00010101000000-000000000000
|
||||
)
|
||||
|
||||
|
|
@ -35,12 +35,11 @@ require (
|
|||
github.com/benbjohnson/clock v1.3.0 // indirect
|
||||
github.com/btcsuite/btcd/btcec/v2 v2.2.0 // indirect
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 // indirect
|
||||
github.com/deepmap/oapi-codegen v1.8.2 // indirect
|
||||
github.com/ethereum/go-ethereum v1.10.21 // indirect
|
||||
github.com/golang/snappy v0.0.4 // indirect
|
||||
github.com/google/go-cmp v0.5.9 // indirect
|
||||
github.com/holiman/uint256 v1.2.1 // indirect
|
||||
github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097 // indirect
|
||||
github.com/inconshreveable/mousetrap v1.1.0 // indirect
|
||||
github.com/klauspost/compress v1.15.11 // indirect
|
||||
github.com/kr/pretty v0.3.0 // indirect
|
||||
github.com/mattn/go-colorable v0.1.13 // indirect
|
||||
|
|
@ -48,6 +47,7 @@ require (
|
|||
github.com/mattn/go-runewidth v0.0.14 // indirect
|
||||
github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect
|
||||
github.com/rivo/uniseg v0.2.0 // indirect
|
||||
github.com/spf13/pflag v1.0.5 // indirect
|
||||
github.com/tidwall/pretty v1.2.1 // indirect
|
||||
github.com/valyala/bytebufferpool v1.0.0 // indirect
|
||||
github.com/valyala/fasthttp v1.41.0 // indirect
|
||||
|
|
@ -60,7 +60,6 @@ require (
|
|||
go.uber.org/goleak v1.1.12 // indirect
|
||||
go.uber.org/multierr v1.8.0 // indirect
|
||||
golang.org/x/crypto v0.1.0 // indirect
|
||||
golang.org/x/net v0.1.0 // indirect
|
||||
golang.org/x/sync v0.1.0 // indirect
|
||||
golang.org/x/sys v0.2.0 // indirect
|
||||
golang.org/x/text v0.4.0 // indirect
|
||||
|
|
|
|||
|
|
@ -32,43 +32,31 @@ github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZx
|
|||
github.com/btcsuite/btcd/btcec/v2 v2.2.0 h1:fzn1qaOt32TuLjFlkzYSsBC35Q3KUjT1SwPxiMSCF5k=
|
||||
github.com/btcsuite/btcd/btcec/v2 v2.2.0/go.mod h1:U7MHm051Al6XmscBQ0BoNydpOTsFAn707034b5nY8zU=
|
||||
github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 h1:q0rUy8C/TYNBQS1+CGKw68tLOFYSNEs0TFnxxnS9+4U=
|
||||
github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
|
||||
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
|
||||
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.0 h1:/8DMNYp9SGi5f0w7uCm6d6M4OU2rGFK09Y2A4Xv7EE0=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 h1:HbphB4TFFXpv7MNrT52FGrrgVXF1owhMVTHFZIlnvd4=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0/go.mod h1:DZGJHZMqrU4JJqFAWUS2UO1+lbSKsdiOoYi9Zzey7Fc=
|
||||
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/ethereum/go-ethereum v1.10.21 h1:5lqsEx92ZaZzRyOqBEXux4/UR06m296RGzN3ol3teJY=
|
||||
github.com/ethereum/go-ethereum v1.10.21/go.mod h1:EYFyF19u3ezGLD4RqOkLq+ZCXzYbLoNDdZlMt7kyKFg=
|
||||
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/gofiber/fiber/v2 v2.40.1 h1:pc7n9VVpGIqNsvg9IPLQhyFEMJL8gCs1kneH5D1pIl4=
|
||||
github.com/gofiber/fiber/v2 v2.40.1/go.mod h1:Gko04sLksnHbzLSRBFWPFdzM9Ws9pRxvvIaohJK1dsk=
|
||||
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.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.4/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.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
|
||||
github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||
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/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/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
|
||||
github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
|
||||
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.4.0 h1:3l4+N6zfMWnkbPEXKng2o2/MR5mSwTrBih4ZEkkz1lg=
|
||||
|
|
@ -84,26 +72,14 @@ 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/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
|
||||
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 h1:bq3VjFmv/sOjHtdEhmkEV4x1AJtvUvOJ2PFAZ5+peKQ=
|
||||
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
|
||||
github.com/mattn/go-runewidth v0.0.14 h1:+xnbZSEeDbOIg5/mE6JF0w6n9duR1l3/WmbinWVwUuU=
|
||||
github.com/mattn/go-runewidth v0.0.14/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
|
||||
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/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 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
|
|
@ -112,14 +88,17 @@ github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
|
|||
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
|
||||
github.com/rogpeppe/go-internal v1.6.1 h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k=
|
||||
github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
|
||||
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
|
||||
github.com/sethvargo/go-envconfig v0.6.0 h1:GxxdoeiNpWgGiVEphNFNObgMYRN/ZvI2dN7rBwadyss=
|
||||
github.com/sethvargo/go-envconfig v0.6.0/go.mod h1:00S1FAhRUuTNJazWBWcJGvEHOM+NO6DhoRMAOX7FY5o=
|
||||
github.com/spf13/cobra v1.7.0 h1:hyqWnYt1ZQShIddO5kBpj3vu05/++x6tJ6dg8EC572I=
|
||||
github.com/spf13/cobra v1.7.0/go.mod h1:uLxZILRyS/50WlhOIKD7W6V5bgeIt+4sICxh6uRMrb0=
|
||||
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/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
|
||||
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
|
||||
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.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
|
|
@ -133,8 +112,6 @@ github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6Kllzaw
|
|||
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
|
||||
github.com/valyala/fasthttp v1.41.0 h1:zeR0Z1my1wDHTRiamBCXVglQdbUwgb9uWG3k1HQz6jY=
|
||||
github.com/valyala/fasthttp v1.41.0/go.mod h1:f6VbjjoI3z1NDOZOv17o6RvtRSWxC77seBFc2uWtgiY=
|
||||
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-20230426150516-e695fad0bed8 h1:rrOyHd+H9a6Op1iUyZNCaI5v9D1syq8jDAYyX/2Q4L8=
|
||||
|
|
@ -161,9 +138,6 @@ go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
|
|||
go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
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-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
|
||||
|
|
@ -173,28 +147,16 @@ golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
|||
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
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-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
|
||||
golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/net v0.0.0-20220906165146-f3363e06e74c/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
|
||||
golang.org/x/net v0.1.0 h1:hZ/3BUoy5aId7sCpA/Tc5lt8DkFgdVS2onTpJsZ/fl0=
|
||||
golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
|
||||
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.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
|
||||
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
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-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
|
|
@ -203,22 +165,17 @@ golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBc
|
|||
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.2.0 h1:ljd4t30dBnAvMZaQCevtY0xLLD0A+bRZXbgLMLU1F/A=
|
||||
golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
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.4.0 h1:BrVqGRd7+k1DiOgtnFvAkoQEWQvBc25ouMJM6429SFg=
|
||||
golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
|
||||
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/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
|
||||
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.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
|
||||
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=
|
||||
|
|
@ -227,9 +184,7 @@ golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8T
|
|||
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/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
|
||||
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.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package infrastructure
|
|||
import (
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/gofiber/fiber/v2/middleware/pprof"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/http/vaa"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/internal/sqs"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.uber.org/zap"
|
||||
|
|
@ -14,7 +15,8 @@ type Server struct {
|
|||
logger *zap.Logger
|
||||
}
|
||||
|
||||
func NewServer(logger *zap.Logger, port string, pprofEnabled bool, isQueueConsumer bool, consumer *sqs.Consumer, db *mongo.Database) *Server {
|
||||
func NewServer(logger *zap.Logger, port string, pprofEnabled bool, isQueueConsumer bool, consumer *sqs.Consumer,
|
||||
db *mongo.Database, vaaController *vaa.Controller) *Server {
|
||||
repository := NewRepository(db, logger)
|
||||
service := NewService(repository, consumer, isQueueConsumer, logger)
|
||||
ctrl := NewController(service, logger)
|
||||
|
|
@ -29,6 +31,8 @@ func NewServer(logger *zap.Logger, port string, pprofEnabled bool, isQueueConsum
|
|||
api.Get("/health", ctrl.HealthCheck)
|
||||
api.Get("/ready", ctrl.ReadyCheck)
|
||||
|
||||
api.Post("/vaa/parse", vaaController.Parse)
|
||||
|
||||
return &Server{
|
||||
app: app,
|
||||
port: port,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,45 @@
|
|||
package vaa
|
||||
|
||||
import (
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/processor"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// Controller definition.
|
||||
type Controller struct {
|
||||
logger *zap.Logger
|
||||
repository *Repository
|
||||
processor processor.ProcessorFunc
|
||||
}
|
||||
|
||||
// NewController creates a Controller instance.
|
||||
func NewController(repository *Repository, processor processor.ProcessorFunc, logger *zap.Logger) *Controller {
|
||||
return &Controller{repository: repository, processor: processor, logger: logger}
|
||||
}
|
||||
|
||||
func (c *Controller) Parse(ctx *fiber.Ctx) error {
|
||||
payload := struct {
|
||||
ID string `json:"id"`
|
||||
}{}
|
||||
|
||||
if err := ctx.BodyParser(&payload); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.logger.Info("Parsing VAA from endpoint", zap.String("id", payload.ID))
|
||||
|
||||
vaa, err := c.repository.FindById(ctx.Context(), payload.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
vaaParsed, err := c.processor(ctx.Context(), vaa.Vaa)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return ctx.JSON(struct {
|
||||
Result any `json:"result"`
|
||||
}{Result: vaaParsed})
|
||||
}
|
||||
|
|
@ -0,0 +1,54 @@
|
|||
package vaa
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type Repository struct {
|
||||
db *mongo.Database
|
||||
logger *zap.Logger
|
||||
vaas *mongo.Collection
|
||||
}
|
||||
|
||||
type VaaDoc struct {
|
||||
ID string `bson:"_id" json:"id"`
|
||||
Vaa []byte `bson:"vaas" json:"vaa"`
|
||||
}
|
||||
|
||||
// NewRepository create a new Repository.
|
||||
func NewRepository(db *mongo.Database, logger *zap.Logger) *Repository {
|
||||
return &Repository{db: db,
|
||||
logger: logger.With(zap.String("module", "VaaRepository")),
|
||||
vaas: db.Collection("vaas"),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Repository) FindById(ctx context.Context, id string) (*VaaDoc, error) {
|
||||
var vaaDoc VaaDoc
|
||||
err := r.vaas.FindOne(ctx, bson.M{"_id": id}).Decode(&vaaDoc)
|
||||
return &vaaDoc, err
|
||||
}
|
||||
|
||||
func (r *Repository) FindPageByTimeRange(ctx context.Context, startTime time.Time, endTime time.Time, page, pageSize int64) ([]*VaaDoc, error) {
|
||||
filter := bson.M{
|
||||
"timestamp": bson.M{
|
||||
"$gte": startTime,
|
||||
"$lt": endTime,
|
||||
},
|
||||
}
|
||||
skip := page * pageSize
|
||||
opts := &options.FindOptions{Skip: &skip, Limit: &pageSize, Sort: bson.M{"timestamp": 1}}
|
||||
cur, err := r.vaas.Find(ctx, filter, opts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var vaas []*VaaDoc
|
||||
err = cur.All(ctx, &vaas)
|
||||
return vaas, err
|
||||
}
|
||||
|
|
@ -2,45 +2,73 @@ package processor
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/wormhole-foundation/wormhole-explorer/parser/parser"
|
||||
"github.com/wormhole-foundation/wormhole/sdk/vaa"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
const (
|
||||
transferPayloadType = 1
|
||||
attestMetaPayloadType = 2
|
||||
transferWithPayloadPayloadType = 3
|
||||
)
|
||||
|
||||
type portalTokenBridgePayload struct {
|
||||
PayloadType int `mapstructure:"payloadType"`
|
||||
Amount *uint64 `mapstructure:"amount"`
|
||||
ToChainID *uint16 `mapstructure:"toChain"`
|
||||
}
|
||||
|
||||
type Processor struct {
|
||||
parser parser.ParserVAAAPIClient
|
||||
repository *parser.Repository
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
func New(repository *parser.Repository, logger *zap.Logger) *Processor {
|
||||
func New(parser parser.ParserVAAAPIClient, repository *parser.Repository, logger *zap.Logger) *Processor {
|
||||
return &Processor{
|
||||
parser: parser,
|
||||
repository: repository,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Processor) Process(ctx context.Context, vaaParsed *parser.ParsedVaaUpdate) error {
|
||||
func (p *Processor) Process(ctx context.Context, vaaBytes []byte) (*parser.ParsedVaaUpdate, error) {
|
||||
|
||||
err := p.repository.UpsertParsedVaa(ctx, *vaaParsed)
|
||||
// unmarshal vaa.
|
||||
vaa, err := vaa.Unmarshal(vaaBytes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// call vaa-payload-parser api to parse a VAA.
|
||||
chainID := uint16(vaa.EmitterChain)
|
||||
emitterAddress := vaa.EmitterAddress.String()
|
||||
sequence := fmt.Sprintf("%d", vaa.Sequence)
|
||||
vaaParseResponse, err := p.parser.Parse(chainID, emitterAddress, sequence, vaa.Payload)
|
||||
if err != nil {
|
||||
if errors.Is(err, parser.ErrInternalError) || errors.Is(err, parser.ErrCallEndpoint) {
|
||||
return nil, err
|
||||
}
|
||||
p.logger.Info("VAA cannot be parsed", zap.Error(err),
|
||||
zap.Uint16("chainID", chainID),
|
||||
zap.String("address", emitterAddress),
|
||||
zap.String("sequence", sequence))
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// create ParsedVaaUpdate to upsert.
|
||||
now := time.Now()
|
||||
vaaParsed := parser.ParsedVaaUpdate{
|
||||
ID: vaa.MessageID(),
|
||||
EmitterChain: chainID,
|
||||
EmitterAddr: emitterAddress,
|
||||
Sequence: sequence,
|
||||
AppID: vaaParseResponse.AppID,
|
||||
Result: vaaParseResponse.Result,
|
||||
Timestamp: vaa.Timestamp,
|
||||
UpdatedAt: &now,
|
||||
}
|
||||
|
||||
err = p.repository.UpsertParsedVaa(ctx, vaaParsed)
|
||||
if err != nil {
|
||||
p.logger.Error("Error inserting vaa in repository",
|
||||
zap.String("id", vaaParsed.ID),
|
||||
zap.Error(err))
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
p.logger.Info("parsed VAA was successfully persisted", zap.String("id", vaaParsed.ID))
|
||||
return nil
|
||||
return &vaaParsed, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,5 +6,5 @@ import (
|
|||
"github.com/wormhole-foundation/wormhole-explorer/parser/parser"
|
||||
)
|
||||
|
||||
// ProcessorFunc is a function to process ParsedVaaUpdate
|
||||
type ProcessorFunc func(context.Context, *parser.ParsedVaaUpdate) error
|
||||
// ProcessorFunc is a function to process vaa message.
|
||||
type ProcessorFunc func(context.Context, []byte) (*parser.ParsedVaaUpdate, error)
|
||||
|
|
|
|||
Loading…
Reference in New Issue