build(deps): bump github.com/VictoriaMetrics/metrics (#40)

Bumps [github.com/VictoriaMetrics/metrics](https://github.com/VictoriaMetrics/metrics) from 1.23.1 to 1.24.0.
- [Commits](https://github.com/VictoriaMetrics/metrics/compare/v1.23.1...v1.24.0)

---
updated-dependencies:
- dependency-name: github.com/VictoriaMetrics/metrics
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

cmd/service/api.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"github.com/VictoriaMetrics/metrics"
+	"github.com/grassrootseconomics/cic-chain-events/pkg/echopprof"
 	"github.com/labstack/echo/v4"
 )
 
@@ -17,5 +18,7 @@ func initApiServer() *echo.Echo {
 		})
 	}
 
+	echopprof.Wrap(server)
+
 	return server
 }

cmd/service/filters.go

@@ -1,31 +1,71 @@
 package main
 
 import (
+	"context"
+	"fmt"
+	"math/big"
 	"strings"
 	"sync"
+	"time"
 
+	"github.com/celo-org/celo-blockchain/common"
+	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-chain-events/internal/filter"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
+	"github.com/grassrootseconomics/w3-celo-patch"
+	"github.com/grassrootseconomics/w3-celo-patch/module/eth"
+	"github.com/grassrootseconomics/w3-celo-patch/w3types"
 )
 
+func initAddressFilter(celoProvider *celoutils.Provider, cache *sync.Map) filter.Filter {
 	var (
-	systemAddress = ko.MustString("chain.system_address")
+		tokenIndexEntryCount big.Int
 	)
 
-func initAddressFilter() filter.Filter {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	// TODO: Bootstrap addresses from smart contract
+	defer cancel()
-	// TODO: Add route to update cache
-	cache := &sync.Map{}
 
-	cache.Store(strings.ToLower(ko.MustString("chain.token_index_address")), "TokenIndex")
+	registryMap, err := celoProvider.RegistryMap(ctx, celoutils.HexToAddress(ko.MustString("chain.registry_address")))
-	cache.Store(strings.ToLower(ko.MustString("chain.gas_faucet_address")), "GasFaucet")
+	if err != nil {
-	cache.Store(strings.ToLower(ko.MustString("chain.user_index_address")), "UserIndex")
+		lo.Fatal("init: critical error creating address filter", "error", err)
+	}
+
+	for k, v := range registryMap {
+		cache.Store(strings.ToLower(v.Hex()), k)
+	}
+
+	if err := celoProvider.Client.CallCtx(
+		ctx,
+		eth.CallFunc(w3.MustNewFunc("entryCount()", "uint256"), registryMap[celoutils.TokenIndex]).Returns(&tokenIndexEntryCount),
+	); err != nil {
+		lo.Fatal("init: critical error creating address filter", "error", err)
+	}
+
+	calls := make([]w3types.Caller, tokenIndexEntryCount.Int64())
+	tokenAddresses := make([]common.Address, tokenIndexEntryCount.Int64())
+
+	entrySig := w3.MustNewFunc("entry(uint256 _idx)", "address")
+
+	// TODO: There is a 5MB limit to a RPC batch call size.
+	// Test if 10k entries will raise an error (future proofed for a lot of years)
+	for i := 0; i < int(tokenIndexEntryCount.Int64()); i++ {
+		calls[i] = eth.CallFunc(entrySig, registryMap[celoutils.TokenIndex], new(big.Int).SetInt64(int64(i))).Returns(&tokenAddresses[i])
+	}
+
+	if err := celoProvider.Client.CallCtx(
+		ctx,
+		calls...,
+	); err != nil {
+		lo.Fatal("init: critical error creating address filter", "error", err)
+	}
+
+	for i, v := range tokenAddresses {
+		cache.Store(strings.ToLower(v.Hex()), fmt.Sprintf("TOKEN_%d", i))
+	}
 
 	return filter.NewAddressFilter(filter.AddressFilterOpts{
 		Cache: cache,
 		Logg:  lo,
-		// TODO: Temporary shortcut
-		SystemAddress: systemAddress,
 	})
 }
 
@@ -41,7 +81,6 @@ func initGasGiftFilter(pub *pub.Pub) filter.Filter {
 	return filter.NewGasFilter(filter.GasFilterOpts{
 		Pub:  pub,
 		Logg: lo,
-		SystemAddress: systemAddress,
 	})
 }
 
@@ -51,3 +90,18 @@ func initRegisterFilter(pub *pub.Pub) filter.Filter {
 		Logg: lo,
 	})
 }
+
+func initApproveFilter(pub *pub.Pub) filter.Filter {
+	return filter.NewApproveFilter(filter.ApproveFilterOpts{
+		Pub:  pub,
+		Logg: lo,
+	})
+}
+
+func initTokenIndexFilter(cache *sync.Map, pub *pub.Pub) filter.Filter {
+	return filter.NewTokenIndexFilter(filter.TokenIndexFilterOpts{
+		Cache: cache,
+		Pub:   pub,
+		Logg:  lo,
+	})
+}

cmd/service/init.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/alitto/pond"
+	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pool"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
 	"github.com/grassrootseconomics/cic-chain-events/internal/store"
@@ -128,3 +129,22 @@ func initPub(natsConn *nats.Conn, jsCtx nats.JetStreamContext) *pub.Pub {
 
 	return pub
 }
+
+func initCeloProvider() *celoutils.Provider {
+	providerOpts := celoutils.ProviderOpts{
+		RpcEndpoint: ko.MustString("chain.rpc_endpoint"),
+	}
+
+	if ko.Bool("chain.testnet") {
+		providerOpts.ChainId = celoutils.TestnetChainId
+	} else {
+		providerOpts.ChainId = celoutils.MainnetChainId
+	}
+
+	provider, err := celoutils.NewProvider(providerOpts)
+	if err != nil {
+		lo.Fatal("init: critical error loading chain provider", "error", err)
+	}
+
+	return provider
+}

cmd/service/main.go

@@ -11,14 +11,13 @@ import (
 	"github.com/grassrootseconomics/cic-chain-events/internal/pipeline"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
 	"github.com/grassrootseconomics/cic-chain-events/internal/syncer"
-	"github.com/knadh/goyesql/v2"
 	"github.com/knadh/koanf/v2"
 	"github.com/labstack/echo/v4"
 	"github.com/zerodha/logf"
 )
 
 type (
-	internalServiceContainer struct {
+	internalServicesContainer struct {
 		apiService *echo.Echo
 		pub        *pub.Pub
 	}
@@ -34,7 +33,6 @@ var (
 
 	ko *koanf.Koanf
 	lo logf.Logger
-	q  goyesql.Queries
 )
 
 func init() {
@@ -57,19 +55,24 @@ func main() {
 	natsConn, jsCtx := initJetStream()
 	jsPub := initPub(natsConn, jsCtx)
 
+	celoProvider := initCeloProvider()
+	cache := &sync.Map{}
+
 	pipeline := pipeline.NewPipeline(pipeline.PipelineOpts{
 		BlockFetcher: graphqlFetcher,
 		Filters: []filter.Filter{
-			initAddressFilter(),
+			initAddressFilter(celoProvider, cache),
 			initGasGiftFilter(jsPub),
 			initTransferFilter(jsPub),
 			initRegisterFilter(jsPub),
+			initApproveFilter(jsPub),
+			initTokenIndexFilter(cache, jsPub),
 		},
 		Logg:  lo,
 		Store: pgStore,
 	})
 
-	internalServices := &internalServiceContainer{
+	internalServices := &internalServicesContainer{
 		pub: jsPub,
 	}
 	syncerStats := &syncer.Stats{}

cmd/service/utils.go

@@ -17,7 +17,7 @@ func createSigChannel() (chan os.Signal, func()) {
 	}
 }
 
-func startGracefulShutdown(ctx context.Context, internalServices *internalServiceContainer) {
+func startGracefulShutdown(ctx context.Context, internalServices *internalServicesContainer) {
 	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 

config.toml

@@ -5,16 +5,15 @@ go_process = true
 # API server
 [service]
 # Host and port
-address = ":5000"
+address = ":5001"
 
 # Geth API endpoints
 [chain]
 graphql_endpoint = ""
 ws_endpoint      = ""
-system_address      = ""
+rpc_endpoint     = ""
-token_index_address = ""
+testnet          = true
-gas_faucet_address  = ""
+registry_address = ""
-user_index_address  = ""
 
 # Syncer configs
 [syncer]

go.mod

@@ -3,19 +3,19 @@ module github.com/grassrootseconomics/cic-chain-events
 go 1.20
 
 require (
-	github.com/VictoriaMetrics/metrics v1.23.1
+	github.com/VictoriaMetrics/metrics v1.24.0
 	github.com/alitto/pond v1.8.3
 	github.com/celo-org/celo-blockchain v1.7.2
-	github.com/grassrootseconomics/celoutils v1.1.1
+	github.com/grassrootseconomics/celoutils v1.4.0
 	github.com/grassrootseconomics/w3-celo-patch v0.2.0
-	github.com/jackc/pgx/v5 v5.3.1
+	github.com/jackc/pgx/v5 v5.4.1
-	github.com/jackc/tern/v2 v2.0.1
+	github.com/jackc/tern/v2 v2.1.0
 	github.com/knadh/goyesql/v2 v2.2.0
 	github.com/knadh/koanf v1.5.0
-	github.com/knadh/koanf/v2 v2.0.0
+	github.com/knadh/koanf/v2 v2.0.1
 	github.com/labstack/echo/v4 v4.10.2
-	github.com/nats-io/nats.go v1.24.0
+	github.com/nats-io/nats.go v1.27.1
-	github.com/stretchr/testify v1.8.2
+	github.com/stretchr/testify v1.8.4
 	github.com/zerodha/logf v0.5.5
 )
 
@@ -54,6 +54,7 @@ require (
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/puddle/v2 v2.2.0 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
+	github.com/klauspost/compress v1.16.5 // indirect
 	github.com/labstack/gommon v0.4.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.17 // indirect
@@ -62,7 +63,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/nats-io/nats-server/v2 v2.9.11 // indirect
-	github.com/nats-io/nkeys v0.3.0 // indirect
+	github.com/nats-io/nkeys v0.4.4 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/onsi/gomega v1.10.1 // indirect
@@ -81,11 +82,11 @@ require (
 	github.com/valyala/fastrand v1.1.0 // indirect
 	github.com/valyala/fasttemplate v1.2.2 // indirect
 	github.com/valyala/histogram v1.2.0 // indirect
-	golang.org/x/crypto v0.6.0 // indirect
+	golang.org/x/crypto v0.9.0 // indirect
-	golang.org/x/net v0.7.0 // indirect
+	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/sys v0.5.0 // indirect
+	golang.org/x/sys v0.8.0 // indirect
-	golang.org/x/text v0.7.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

go.sum

@@ -48,8 +48,8 @@ github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 h1:fLjPD/aNc3UIO
 github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
 github.com/VictoriaMetrics/fastcache v1.6.0 h1:C/3Oi3EiBCqufydp1neRZkqcwmEiuRT9c3fqvvgKm5o=
 github.com/VictoriaMetrics/fastcache v1.6.0/go.mod h1:0qHz5QP0GMX4pfmMA/zt5RgfNuXJrTP0zS7DqpHGGTw=
-github.com/VictoriaMetrics/metrics v1.23.1 h1:/j8DzeJBxSpL2qSIdqnRFLvQQhbJyJbbEi22yMm7oL0=
+github.com/VictoriaMetrics/metrics v1.24.0 h1:ILavebReOjYctAGY5QU2F9X0MYvkcrG3aEn2RKa1Zkw=
-github.com/VictoriaMetrics/metrics v1.23.1/go.mod h1:rAr/llLpEnAdTehiNlUxKgnjcOuROSzpw0GvjpEbvFc=
+github.com/VictoriaMetrics/metrics v1.24.0/go.mod h1:eFT25kvsTidQFHb6U0oa0rTrDRdz4xTYjpL8+UPohys=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
 github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -267,8 +267,8 @@ github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB7
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/graph-gophers/graphql-go v1.3.0/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
-github.com/grassrootseconomics/celoutils v1.1.1 h1:REsndvfBkPN8UKOoQFNEGm/sCwKtTm+woYtgMl3bfZ0=
+github.com/grassrootseconomics/celoutils v1.4.0 h1:AJNKiOpfnQqZ3kRxeUlhWH/zlDDjhtbs/OzAMb5zU4A=
-github.com/grassrootseconomics/celoutils v1.1.1/go.mod h1:Uo5YRy6AGLAHDZj9jaOI+AWoQ1H3L0v79728pPMkm9Q=
+github.com/grassrootseconomics/celoutils v1.4.0/go.mod h1:Uo5YRy6AGLAHDZj9jaOI+AWoQ1H3L0v79728pPMkm9Q=
 github.com/grassrootseconomics/w3-celo-patch v0.2.0 h1:YqibbPzX0tQKmxU1nUGzThPKk/fiYeYZY6Aif3eyu8U=
 github.com/grassrootseconomics/w3-celo-patch v0.2.0/go.mod h1:WhBXNzNIvHmS6B2hAeShs56oa9Azb4jQSrOMKuMdBWw=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -346,12 +346,12 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.3.1 h1:Fcr8QJ1ZeLi5zsPZqQeUZhNhxfkkKBOgJuYkJHoBOtU=
+github.com/jackc/pgx/v5 v5.4.1 h1:oKfB/FhuVtit1bBM3zNRRsZ925ZkMN3HXL+LgLUM9lE=
-github.com/jackc/pgx/v5 v5.3.1/go.mod h1:t3JDKnCBlYIc0ewLF0Q7B8MXmoIaBOZj/ic7iHozM/8=
+github.com/jackc/pgx/v5 v5.4.1/go.mod h1:q6iHT8uDNXWiFNOlRqJzBTaSH3+2xCXkokxHZC5qWFY=
 github.com/jackc/puddle/v2 v2.2.0 h1:RdcDk92EJBuBS55nQMMYFXTxwstHug4jkhT5pq8VxPk=
 github.com/jackc/puddle/v2 v2.2.0/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
-github.com/jackc/tern/v2 v2.0.1 h1:2J05jlmYFsNQe9rGsqoNs0+6eDm3iJns8RQb6DJl1V8=
+github.com/jackc/tern/v2 v2.1.0 h1:yqx1rppJY0WfDC7nAmRCLODRrdlLWO8VspuTD88nX7k=
-github.com/jackc/tern/v2 v2.0.1/go.mod h1:4cpqN/grjWYeRWcKXah5YGoviJKJuoqNLoORKLumoG0=
+github.com/jackc/tern/v2 v2.1.0/go.mod h1:4cpqN/grjWYeRWcKXah5YGoviJKJuoqNLoORKLumoG0=
 github.com/jackpal/go-nat-pmp v1.0.2-0.20160603034137-1fa385a6f458/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
 github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus=
 github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
@@ -380,7 +380,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
 github.com/klauspost/compress v1.4.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
-github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B9Sx9c=
+github.com/klauspost/compress v1.16.5 h1:IFV2oUNUzZaz+XyusxpLzpzS8Pt5rh0Z16For/djlyI=
+github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/klauspost/cpuid v0.0.0-20170728055534-ae7887de9fa5/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/crc32 v0.0.0-20161016154125-cb6bfca970f6/go.mod h1:+ZoRqAPRLkC4NPOvfYeR5KNOrY6TD+/sAC3HXPZgDYg=
 github.com/klauspost/pgzip v1.0.2-0.20170402124221-0bf5dcad4ada/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
@@ -388,8 +389,8 @@ github.com/knadh/goyesql/v2 v2.2.0 h1:DNQIzgITmMTXA+z+jDzbXCpgr7fGD6Hp0AJ7ZLEAem
 github.com/knadh/goyesql/v2 v2.2.0/go.mod h1:is+wK/XQBukYK3DdKfpJRyDH9U/ZTMyX2u6DFijjRnI=
 github.com/knadh/koanf v1.5.0 h1:q2TSd/3Pyc/5yP9ldIrSdIz26MCcyNQzW0pEAugLPNs=
 github.com/knadh/koanf v1.5.0/go.mod h1:Hgyjp4y8v44hpZtPzs7JZfRAW5AhN7KfZcwv1RYggDs=
-github.com/knadh/koanf/v2 v2.0.0 h1:XPQ5ilNnwnNaHrfQ1YpTVhUAjcGHnEKA+lRpipQv02Y=
+github.com/knadh/koanf/v2 v2.0.1 h1:1dYGITt1I23x8cfx8ZnldtezdyaZtfAuRtIFOiRzK7g=
-github.com/knadh/koanf/v2 v2.0.0/go.mod h1:ZeiIlIDXTE7w1lMT6UVcNiRAS2/rCeLn/GdLNvY1Dus=
+github.com/knadh/koanf/v2 v2.0.1/go.mod h1:ZeiIlIDXTE7w1lMT6UVcNiRAS2/rCeLn/GdLNvY1Dus=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
@@ -475,10 +476,10 @@ github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416/go.mod h1:NBIhNtsFMo
 github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
 github.com/nats-io/nats-server/v2 v2.9.11 h1:4y5SwWvWI59V5mcqtuoqKq6L9NDUydOP3Ekwuwl8cZI=
 github.com/nats-io/nats-server/v2 v2.9.11/go.mod h1:b0oVuxSlkvS3ZjMkncFeACGyZohbO4XhSqW1Lt7iRRY=
-github.com/nats-io/nats.go v1.24.0 h1:CRiD8L5GOQu/DcfkmgBcTTIQORMwizF+rPk6T0RaHVQ=
+github.com/nats-io/nats.go v1.27.1 h1:OuYnal9aKVSnOzLQIzf7554OXMCG7KbaTkCSBHRcSoo=
-github.com/nats-io/nats.go v1.24.0/go.mod h1:dVQF+BK3SzUZpwyzHedXsvH3EO38aVKuOPkkHlv5hXA=
+github.com/nats-io/nats.go v1.27.1/go.mod h1:XpbWUlOElGwTYbMR7imivs7jJj9GtK7ypv321Wp6pjc=
-github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
+github.com/nats-io/nkeys v0.4.4 h1:xvBJ8d69TznjcQl9t6//Q5xXuVhyYiSos6RPtvQNTwA=
-github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
+github.com/nats-io/nkeys v0.4.4/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/npillmayer/nestext v0.1.3/go.mod h1:h2lrijH8jpicr25dFY+oAJLyzlya6jhnuG+zWp9L0Uk=
@@ -576,18 +577,14 @@ github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnIn
 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.1.1/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.2.0/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 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.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
-github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
-github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
-github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 h1:epCh84lMvA70Z7CTTCmYQn2CKbY8j86K7/FAIr141uY=
 github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7/go.mod h1:q4W45IWZaF22tdD+VEXcAWRA037jwmWEB5VWYORlTpc=
 github.com/tinylib/msgp v1.0.2/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
@@ -639,12 +636,11 @@ golang.org/x/crypto v0.0.0-20200115085410-6d4e4cb37c7d/go.mod h1:LzIPMQfyMNhhGPh
 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-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.3.0/go.mod h1:hebNnKkNXi2UzZN1eVRvBB7co0a+JxK6XbPiWVs/3J4=
-golang.org/x/crypto v0.6.0 h1:qfktjS5LUO+fFKeJXZ+ikTRijMmljikvG68fpMMruSc=
+golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
-golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
+golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -706,8 +702,8 @@ golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod h1:9tjilg8BloeKEkVJvy
 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.2.0/go.mod h1:KqCZLdyyvdV855qA2rE3GC2aiw5xGR5TEjj8smXukLY=
-golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
+golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
-golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -788,8 +784,8 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc
 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.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
+golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
-golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.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=
@@ -804,8 +800,8 @@ 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/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
+golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
-golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20201208040808-7e3f01d25324/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

internal/filter/address_filter.go

@@ -12,13 +12,11 @@ type (
 	AddressFilterOpts struct {
 		Cache *sync.Map
 		Logg  logf.Logger
-		SystemAddress string
 	}
 
 	AddressFilter struct {
 		cache *sync.Map
 		logg  logf.Logger
-		systemAddress string
 	}
 )
 
@@ -26,15 +24,10 @@ func NewAddressFilter(o AddressFilterOpts) Filter {
 	return &AddressFilter{
 		cache: o.Cache,
 		logg:  o.Logg,
-		systemAddress: o.SystemAddress,
 	}
 }
 
 func (f *AddressFilter) Execute(_ context.Context, transaction *fetch.Transaction) (bool, error) {
-	if transaction.From.Address == f.systemAddress {
-		return true, nil
-	}
-
 	if _, found := f.cache.Load(transaction.To.Address); found {
 		return true, nil
 	}

internal/filter/approve_filter.go

@@ -0,0 +1,85 @@
+package filter
+
+import (
+	"context"
+	"math/big"
+
+	"github.com/celo-org/celo-blockchain/common"
+	"github.com/celo-org/celo-blockchain/common/hexutil"
+	"github.com/grassrootseconomics/celoutils"
+	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
+	"github.com/grassrootseconomics/cic-chain-events/pkg/fetch"
+	"github.com/grassrootseconomics/w3-celo-patch"
+	"github.com/zerodha/logf"
+)
+
+type (
+	ApproveFilterOpts struct {
+		Logg logf.Logger
+		Pub  *pub.Pub
+	}
+
+	ApproveFilter struct {
+		logg logf.Logger
+		pub  *pub.Pub
+	}
+)
+
+const (
+	approveEventSubject = "CHAIN.approve"
+)
+
+var (
+	approveSig = w3.MustNewFunc("approve(address, uint256)", "bool")
+)
+
+func NewApproveFilter(o ApproveFilterOpts) Filter {
+	return &ApproveFilter{
+		logg: o.Logg,
+		pub:  o.Pub,
+	}
+}
+
+func (f *ApproveFilter) Execute(_ context.Context, transaction *fetch.Transaction) (bool, error) {
+	if len(transaction.InputData) < 10 {
+		return true, nil
+	}
+
+	if transaction.InputData[:10] == "0x095ea7b3" {
+		var (
+			address common.Address
+			value   big.Int
+		)
+
+		if err := approveSig.DecodeArgs(w3.B(transaction.InputData), &address, &value); err != nil {
+			return false, err
+		}
+
+		approveEvent := &pub.MinimalTxInfo{
+			Block:           transaction.Block.Number,
+			ContractAddress: celoutils.ChecksumAddress(transaction.To.Address),
+			Timestamp:       hexutil.MustDecodeUint64(transaction.Block.Timestamp),
+			From:            celoutils.ChecksumAddress(transaction.From.Address),
+			To:              address.Hex(),
+			TxHash:          transaction.Hash,
+			TxIndex:         transaction.Index,
+			TXType:          "approve",
+		}
+
+		if transaction.Status == 1 {
+			approveEvent.Success = true
+		}
+
+		if err := f.pub.Publish(
+			approveEventSubject,
+			transaction.Hash,
+			approveEvent,
+		); err != nil {
+			return false, err
+		}
+
+		return true, nil
+	}
+
+	return true, nil
+}

internal/filter/gas_filter.go

@@ -3,28 +3,32 @@ package filter
 import (
 	"context"
 
+	"github.com/celo-org/celo-blockchain/common"
 	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
 	"github.com/grassrootseconomics/cic-chain-events/pkg/fetch"
+	"github.com/grassrootseconomics/w3-celo-patch"
 	"github.com/zerodha/logf"
 )
 
 const (
-	gasFilterEventSubject = "CHAIN.gas"
+	gasEventSubject = "CHAIN.gas"
+)
+
+var (
+	giveToSig = w3.MustNewFunc("giveTo(address)", "uint256")
 )
 
 type (
 	GasFilterOpts struct {
 		Logg logf.Logger
 		Pub  *pub.Pub
-		SystemAddress string
 	}
 
 	GasFilter struct {
 		logg logf.Logger
 		pub  *pub.Pub
-		systemAddress string
 	}
 )
 
@@ -32,34 +36,39 @@ func NewGasFilter(o GasFilterOpts) Filter {
 	return &GasFilter{
 		logg: o.Logg,
 		pub:  o.Pub,
-		systemAddress: o.SystemAddress,
 	}
 }
 
 func (f *GasFilter) Execute(_ context.Context, transaction *fetch.Transaction) (bool, error) {
-	transferValue, err := hexutil.DecodeUint64(transaction.Value)
+	if len(transaction.InputData) < 10 {
-	if err != nil {
+		return true, nil
+	}
+
+	if transaction.InputData[:10] == "0x63e4bff4" {
+		var address common.Address
+
+		if err := giveToSig.DecodeArgs(w3.B(transaction.InputData), &address); err != nil {
 			return false, err
 		}
 
-	// TODO: This is a temporary shortcut to gift gas. Switch to gas faucet contract.
+		giveToEvent := &pub.MinimalTxInfo{
-	if transaction.From.Address == f.systemAddress && transferValue > 0 {
-		transferEvent := &pub.MinimalTxInfo{
 			Block:           transaction.Block.Number,
-			To:      celoutils.ChecksumAddress(transaction.To.Address),
+			ContractAddress: celoutils.ChecksumAddress(transaction.To.Address),
+			Timestamp:       hexutil.MustDecodeUint64(transaction.Block.Timestamp),
+			To:              address.Hex(),
 			TxHash:          transaction.Hash,
 			TxIndex:         transaction.Index,
-			Value:   transferValue,
+			TXType:          "gas",
 		}
 
 		if transaction.Status == 1 {
-			transferEvent.Success = true
+			giveToEvent.Success = true
 		}
 
 		if err := f.pub.Publish(
-			gasFilterEventSubject,
+			gasEventSubject,
 			transaction.Hash,
-			transferEvent,
+			giveToEvent,
 		); err != nil {
 			return false, err
 		}

internal/filter/register_filter.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/celo-org/celo-blockchain/common"
+	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
 	"github.com/grassrootseconomics/cic-chain-events/pkg/fetch"
@@ -16,7 +17,7 @@ const (
 )
 
 var (
-	addSig = w3.MustNewFunc("add(address)", "bool")
+	registerSig = w3.MustNewFunc("register(address)", "")
 )
 
 type (
@@ -43,19 +44,21 @@ func (f *RegisterFilter) Execute(_ context.Context, transaction *fetch.Transacti
 		return true, nil
 	}
 
-	if transaction.InputData[:10] == "0x0a3b0a4f" {
+	if transaction.InputData[:10] == "0x4420e486" {
 		var address common.Address
 
-		if err := addSig.DecodeArgs(w3.B(transaction.InputData), &address); err != nil {
+		if err := registerSig.DecodeArgs(w3.B(transaction.InputData), &address); err != nil {
 			return false, err
 		}
 
 		addEvent := &pub.MinimalTxInfo{
 			Block:           transaction.Block.Number,
-			ContractAddress: transaction.To.Address,
+			ContractAddress: celoutils.ChecksumAddress(transaction.To.Address),
-			To:              celoutils.ChecksumAddress(transaction.To.Address),
+			Timestamp:       hexutil.MustDecodeUint64(transaction.Block.Timestamp),
+			To:              address.Hex(),
 			TxHash:          transaction.Hash,
 			TxIndex:         transaction.Index,
+			TXType:          "register",
 		}
 
 		if transaction.Status == 1 {

internal/filter/token_index_filter.go

@@ -0,0 +1,84 @@
+package filter
+
+import (
+	"context"
+	"strings"
+	"sync"
+
+	"github.com/celo-org/celo-blockchain/common"
+	"github.com/celo-org/celo-blockchain/common/hexutil"
+	"github.com/grassrootseconomics/celoutils"
+	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
+	"github.com/grassrootseconomics/cic-chain-events/pkg/fetch"
+	"github.com/grassrootseconomics/w3-celo-patch"
+	"github.com/zerodha/logf"
+)
+
+const (
+	tokenIndexFilterEventSubject = "CHAIN.tokenAdded"
+)
+
+var (
+	addSig = w3.MustNewFunc("add(address)", "bool")
+)
+
+type (
+	TokenIndexFilterOpts struct {
+		Cache *sync.Map
+		Logg  logf.Logger
+		Pub   *pub.Pub
+	}
+
+	TokenIndexFilter struct {
+		pub   *pub.Pub
+		cache *sync.Map
+		logg  logf.Logger
+	}
+)
+
+func NewTokenIndexFilter(o TokenIndexFilterOpts) Filter {
+	return &TokenIndexFilter{
+		cache: o.Cache,
+		logg:  o.Logg,
+		pub:   o.Pub,
+	}
+}
+
+func (f *TokenIndexFilter) Execute(_ context.Context, transaction *fetch.Transaction) (bool, error) {
+	if len(transaction.InputData) < 10 {
+		return true, nil
+	}
+
+	if transaction.InputData[:10] == "0x0a3b0a4f" {
+		var address common.Address
+
+		if err := addSig.DecodeArgs(w3.B(transaction.InputData), &address); err != nil {
+			return false, err
+		}
+
+		f.cache.Store(strings.ToLower(address.Hex()), transaction.Hash)
+
+		addEvent := &pub.MinimalTxInfo{
+			Block:           transaction.Block.Number,
+			ContractAddress: celoutils.ChecksumAddress(transaction.To.Address),
+			Timestamp:       hexutil.MustDecodeUint64(transaction.Block.Timestamp),
+			To:              address.Hex(),
+			TxHash:          transaction.Hash,
+			TxIndex:         transaction.Index,
+			TXType:          "tokenAdd",
+		}
+
+		if transaction.Status == 1 {
+			addEvent.Success = true
+		}
+
+		if err := f.pub.Publish(
+			tokenIndexFilterEventSubject,
+			transaction.Hash,
+			addEvent,
+		); err != nil {
+			return false, err
+		}
+	}
+	return true, nil
+}

internal/filter/transfer_filter.go

@@ -5,6 +5,7 @@ import (
 	"math/big"
 
 	"github.com/celo-org/celo-blockchain/common"
+	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pub"
 	"github.com/grassrootseconomics/cic-chain-events/pkg/fetch"
@@ -57,15 +58,15 @@ func (f *TransferFilter) Execute(_ context.Context, transaction *fetch.Transacti
 			return false, err
 		}
 
-		f.logg.Debug("transfer_filter: new reg", "transfer", to)
-
 		transferEvent := &pub.MinimalTxInfo{
 			Block:           transaction.Block.Number,
 			From:            celoutils.ChecksumAddress(transaction.From.Address),
+			Timestamp:       hexutil.MustDecodeUint64(transaction.Block.Timestamp),
 			To:              to.Hex(),
-			ContractAddress: transaction.To.Address,
+			ContractAddress: celoutils.ChecksumAddress(transaction.To.Address),
 			TxHash:          transaction.Hash,
 			TxIndex:         transaction.Index,
+			TXType:          "transfer",
 			Value:           value.Uint64(),
 		}
 
@@ -93,15 +94,15 @@ func (f *TransferFilter) Execute(_ context.Context, transaction *fetch.Transacti
 			return false, err
 		}
 
-		f.logg.Debug("transfer_filter: new reg", "transferFrom", to)
-
 		transferEvent := &pub.MinimalTxInfo{
 			Block:           transaction.Block.Number,
 			From:            from.Hex(),
+			Timestamp:       hexutil.MustDecodeUint64(transaction.Block.Timestamp),
 			To:              to.Hex(),
-			ContractAddress: transaction.To.Address,
+			ContractAddress: celoutils.ChecksumAddress(transaction.To.Address),
 			TxHash:          transaction.Hash,
 			TxIndex:         transaction.Index,
+			TXType:          "transferFrom",
 			Value:           value.Uint64(),
 		}
 
@@ -128,15 +129,15 @@ func (f *TransferFilter) Execute(_ context.Context, transaction *fetch.Transacti
 			return false, err
 		}
 
-		f.logg.Debug("transfer_filter: new reg", "mintTo", to)
-
 		transferEvent := &pub.MinimalTxInfo{
 			Block:           transaction.Block.Number,
 			From:            celoutils.ChecksumAddress(transaction.From.Address),
+			Timestamp:       hexutil.MustDecodeUint64(transaction.Block.Timestamp),
 			To:              to.Hex(),
-			ContractAddress: transaction.To.Address,
+			ContractAddress: celoutils.ChecksumAddress(transaction.To.Address),
 			TxHash:          transaction.Hash,
 			TxIndex:         transaction.Index,
+			TXType:          "mintTo",
 			Value:           value.Uint64(),
 		}
 

internal/pub/jetstream.go

@@ -31,8 +31,10 @@ type (
 		To              string `json:"to"`
 		ContractAddress string `json:"contractAddress"`
 		Success         bool   `json:"success"`
+		Timestamp       uint64 `json:"timestamp"`
 		TxHash          string `json:"transactionHash"`
 		TxIndex         uint   `json:"transactionIndex"`
+		TXType          string `json:"txType"`
 		Value           uint64 `json:"value"`
 	}
 )

internal/syncer/head.go

@@ -7,12 +7,14 @@ import (
 	"github.com/alitto/pond"
 	"github.com/celo-org/celo-blockchain/core/types"
 	"github.com/celo-org/celo-blockchain/ethclient"
+	"github.com/celo-org/celo-blockchain/event"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pipeline"
 	"github.com/zerodha/logf"
 )
 
 const (
 	jobTimeout         = 5 * time.Second
+	resubscribeBackoff = 2 * time.Second
 )
 
 type (
@@ -53,10 +55,13 @@ func NewHeadSyncer(o HeadSyncerOpts) (*HeadSyncer, error) {
 func (hs *HeadSyncer) Start(ctx context.Context) error {
 	headerReceiver := make(chan *types.Header, 1)
 
-	sub, err := hs.ethClient.SubscribeNewHead(context.Background(), headerReceiver)
+	sub := event.ResubscribeErr(resubscribeBackoff, func(ctx context.Context, err error) (event.Subscription, error) {
 		if err != nil {
-		return err
+			hs.logg.Error("head syncer: resubscribe error", "error", err)
 		}
+
+		return hs.ethClient.SubscribeNewHead(ctx, headerReceiver)
+	})
 	defer sub.Unsubscribe()
 
 	for {
@@ -64,8 +69,6 @@ func (hs *HeadSyncer) Start(ctx context.Context) error {
 		case <-ctx.Done():
 			hs.logg.Info("head syncer: shutdown signal received")
 			return nil
-		case err := <-sub.Err():
-			return err
 		case header := <-headerReceiver:
 			blockNumber := header.Number.Uint64()
 			hs.logg.Debug("head syncer: received new block", "block", blockNumber)
@@ -75,7 +78,7 @@ func (hs *HeadSyncer) Start(ctx context.Context) error {
 				defer cancel()
 
 				if err := hs.pipeline.Run(ctx, blockNumber); err != nil {
-					hs.logg.Error("head syncer: piepline run error", "error", err)
+					hs.logg.Error("head syncer: pipeline run error", "error", err)
 				}
 			})
 		}

pkg/echopprof/echopprof.go

@@ -0,0 +1,158 @@
+// MIT License
+
+// Copyright (c) 2017 Leslie Zheng
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package echopprof
+
+import (
+	"net/http/pprof"
+	"strings"
+
+	"github.com/labstack/echo/v4"
+)
+
+// Wrap adds several routes from package `net/http/pprof` to *echo.Echo object.
+func Wrap(e *echo.Echo) {
+	WrapGroup("", e.Group("/debug/pprof"))
+}
+
+// Wrapper make sure we are backward compatible.
+var Wrapper = Wrap
+
+// WrapGroup adds several routes from package `net/http/pprof` to *echo.Group object.
+func WrapGroup(prefix string, g *echo.Group) {
+	routers := []struct {
+		Method  string
+		Path    string
+		Handler echo.HandlerFunc
+	}{
+		{"GET", "", IndexHandler()},
+		{"GET", "/", IndexHandler()},
+		{"GET", "/heap", HeapHandler()},
+		{"GET", "/goroutine", GoroutineHandler()},
+		{"GET", "/block", BlockHandler()},
+		{"GET", "/threadcreate", ThreadCreateHandler()},
+		{"GET", "/cmdline", CmdlineHandler()},
+		{"GET", "/profile", ProfileHandler()},
+		{"GET", "/symbol", SymbolHandler()},
+		{"POST", "/symbol", SymbolHandler()},
+		{"GET", "/trace", TraceHandler()},
+		{"GET", "/mutex", MutexHandler()},
+		{"GET", "/allocs", AllocsHandler()},
+	}
+
+	for _, r := range routers {
+		switch r.Method {
+		case "GET":
+			g.GET(strings.TrimPrefix(r.Path, prefix), r.Handler)
+		case "POST":
+			g.POST(strings.TrimPrefix(r.Path, prefix), r.Handler)
+		}
+	}
+}
+
+// IndexHandler will pass the call from /debug/pprof to pprof.
+func IndexHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Index(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// HeapHandler will pass the call from /debug/pprof/heap to pprof.
+func HeapHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Handler("heap").ServeHTTP(ctx.Response(), ctx.Request())
+		return nil
+	}
+}
+
+// GoroutineHandler will pass the call from /debug/pprof/goroutine to pprof.
+func GoroutineHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Handler("goroutine").ServeHTTP(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// BlockHandler will pass the call from /debug/pprof/block to pprof.
+func BlockHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Handler("block").ServeHTTP(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// ThreadCreateHandler will pass the call from /debug/pprof/threadcreate to pprof.
+func ThreadCreateHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Handler("threadcreate").ServeHTTP(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// CmdlineHandler will pass the call from /debug/pprof/cmdline to pprof.
+func CmdlineHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Cmdline(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// ProfileHandler will pass the call from /debug/pprof/profile to pprof.
+func ProfileHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Profile(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// SymbolHandler will pass the call from /debug/pprof/symbol to pprof.
+func SymbolHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Symbol(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// TraceHandler will pass the call from /debug/pprof/trace to pprof.
+func TraceHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Trace(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// MutexHandler will pass the call from /debug/pprof/mutex to pprof.
+func MutexHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Handler("mutex").ServeHTTP(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}
+
+// AllocsHandler will pass the call from /debug/pprof/allocs to pprof.
+func AllocsHandler() echo.HandlerFunc {
+	return func(ctx echo.Context) error {
+		pprof.Handler("allocs").ServeHTTP(ctx.Response().Writer, ctx.Request())
+		return nil
+	}
+}

pkg/fetch/graphql.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"time"
 )
@@ -49,13 +50,18 @@ func (f *Graphql) Block(ctx context.Context, blockNumber uint64) (FetchResponse,
 	if err != nil {
 		return fetchResponse, err
 	}
-	defer resp.Body.Close()
 
 	if resp.StatusCode >= http.StatusBadRequest {
 		return fetchResponse, fmt.Errorf("error fetching block %s", resp.Status)
 	}
 
-	if err := json.NewDecoder(resp.Body).Decode(&fetchResponse); err != nil {
+	out, err := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if err != nil {
+		return fetchResponse, nil
+	}
+
+	if err := json.Unmarshal(out, &fetchResponse); err != nil {
 		return fetchResponse, err
 	}