feat: temporaily disable cache

.github/dependabot.yaml

@@ -1,6 +0,0 @@
-version: 2
-updates:
-  - package-ecosystem: "gomod"
-    directory: "/"
-    schedule:
-      interval: "weekly"

.github/workflows/docker.yaml

@@ -1,56 +0,0 @@
-name: release
-
-on:
-  push:
-    tags:
-      - "v*"
-
-jobs:
-  docker:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v2
-
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v2
-
-      - name: Check out repo
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: 0
-
-      - name: Cache Docker layers
-        uses: actions/cache@v3
-        with:
-          path: /tmp/.buildx-cache
-          key: ${{ runner.os }}-buildx-${{ github.sha }}
-          restore-keys: |
-            ${{ runner.os }}-buildx-
-
-      - name: Login to GHCR Docker registry
-        uses: docker/login-action@v1
-        with:
-          registry: ghcr.io
-          username: ${{ github.repository_owner }}
-          password: ${{ secrets.GITHUB_TOKEN }}
-
-      - name: Set outputs
-        run: |
-          echo "RELEASE_TAG=${GITHUB_REF#refs/*/}" >> $GITHUB_ENV \
-          && echo "RELEASE_SHORT_COMMIT=$(git rev-parse --short HEAD)" >> $GITHUB_ENV
-
-      - name: Build and push image
-        uses: docker/build-push-action@v2
-        with:
-          context: ./
-          file: ./Dockerfile
-          platforms: linux/amd64
-          push: true
-          build-args: |
-            BUILD=${{ env.RELEASE_SHORT_COMMIT }}
-          cache-from: type=local,src=/tmp/.buildx-cache
-          cache-to: type=local,dest=/tmp/.buildx-cache
-          tags: |
-            ghcr.io/grassrootseconomics/celo-indexer:latest
-            ghcr.io/grassrootseconomics/celo-indexer:${{ env.RELEASE_TAG }}

Dockerfile

@@ -1,17 +1,18 @@
 FROM golang:1.22.3-bookworm as build
 
-ENV CGO_ENABLED=1
+# Should be 0 unless we import celo-blockchain or related libraries
+ENV CGO_ENABLED=0
 
 ARG BUILDPLATFORM
 ARG TARGETPLATFORM
 ARG BUILD=dev
 
 RUN echo "Building on $BUILDPLATFORM, building for $TARGETPLATFORM"
-WORKDIR /code
+WORKDIR /build
 
 COPY . .
 RUN go mod download
-RUN go build -o /build/celo-indexer -ldflags="-X main.build=${BUILD} -s -w" cmd/service/main.go
+RUN go build -o celo-indexer -ldflags="-X main.build=${BUILD} -s -w" cmd/*
 
 FROM debian:bookworm-slim
 
@@ -21,11 +22,7 @@ WORKDIR /service
 
 COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 COPY --from=build /build/* .
-COPY migrations migrations/
-COPY config.toml .
-COPY queries.sql .
-COPY LICENSE .
 
-EXPOSE 5002
+EXPOSE 5001
 
 CMD ["./celo-indexer"]

README.md

@@ -1,61 +0,0 @@
-# celo-indexer
-
-![GitHub Tag](https://img.shields.io/github/v/tag/grassrootseconomics/celo-indexer)
-
-A lightweight Postgres chain indexer designed to couple with [celo-tracker](https://github.com/grassrootseconomics/celo-tracker) to index all relevant GE related blockchain data on the Celo blockchain.
-
-## Getting Started
-
-### Prerequisites
-
-* Git
-* Docker
-* Postgres server
-* Access to a `celo-tracker` instance
-
-See [docker-compose.yaml](dev/docker-compose.yaml) for an example on how to run and deploy a single instance.
-
-### 1. Build the Docker image
-
-We provide pre-built images for `linux/amd64`. See the packages tab on Github.
-
-If you are on any other platform:
-
-```bash
-git clone https://github.com/grassrootseconomics/celo-indexer.git
-cd celo-indexer
-docker buildx build --build-arg BUILD=$(git rev-parse --short HEAD) --tag celo-indexer:$(git rev-parse --short HEAD) --tag celo-indexer:latest .
-docker images
-```
-
-### 2. Run Postgres
-
-For an example, see `dev/docker-compose.postgres.yaml`.
-
-### 3. Update config values
-
-See `.env.example` on how to override default values defined in `config.toml` using env variables. Alternatively, mount your own config.toml either during build time or Docker runtime.
-
-```bash
-# Override only specific config values
-nano .env.example
-mv .env.example .env
-```
-
-Special env variables:
-
-* DEV=*
-
-Refer to [`config.toml`](config.toml) to understand different config value settings.
-
-
-### 4. Run the indexer
-
-```bash
-cd dev
-docker compose up
-```
-
-## License
-
-[AGPL-3.0](LICENSE).

cmd/bootstrap/main.go

@@ -1,244 +0,0 @@
-package main
-
-import (
-	"context"
-	"flag"
-	"log/slog"
-	"os"
-	"time"
-
-	"github.com/celo-org/celo-blockchain/common"
-	"github.com/grassrootseconomics/celo-indexer/internal/util"
-	"github.com/grassrootseconomics/celoutils/v3"
-	"github.com/grassrootseconomics/w3-celo"
-	"github.com/grassrootseconomics/w3-celo/module/eth"
-	"github.com/jackc/pgx/v5/pgxpool"
-	"github.com/knadh/koanf/v2"
-)
-
-type TokenArgs struct {
-	ContractAddress string
-	TokenName       string
-	TokenSymbol     string
-	TokenDecimals   uint8
-}
-
-const (
-	insertTokenQuery = `INSERT INTO tokens(
-		contract_address,
-		token_name,
-		token_symbol,
-		token_decimals
-	) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING`
-)
-
-var (
-	build = "dev"
-
-	confFlag string
-
-	lo *slog.Logger
-	ko *koanf.Koanf
-
-	dbPool *pgxpool.Pool
-)
-
-func init() {
-	flag.StringVar(&confFlag, "config", "config.toml", "Config file location")
-	flag.Parse()
-
-	lo = util.InitLogger()
-	ko = util.InitConfig(lo, confFlag)
-
-	lo.Info("starting GE indexer token bootstrapper", "build", build)
-}
-
-func main() {
-	var (
-		tokenRegistryGetter = w3.MustNewFunc("tokenRegistry()", "address")
-		nameGetter          = w3.MustNewFunc("name()", "string")
-		symbolGetter        = w3.MustNewFunc("symbol()", "string")
-		decimalsGetter      = w3.MustNewFunc("decimals()", "uint8")
-	)
-
-	chainProvider := celoutils.NewProvider(ko.MustString("chain.rpc_endpoint"), ko.MustInt64("chain.chainid"))
-
-	var err error
-	dbPool, err = newPgStore()
-	if err != nil {
-		lo.Error("could not initialize postgres store", "error", err)
-		os.Exit(1)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
-	defer cancel()
-
-	for _, registry := range ko.MustStrings("bootstrap.ge_registries") {
-		registryMap, err := chainProvider.RegistryMap(ctx, celoutils.HexToAddress(registry))
-		if err != nil {
-			lo.Error("could not fetch registry", "error", err)
-			os.Exit(1)
-		}
-
-		if tokenIndex := registryMap[celoutils.TokenIndex]; tokenIndex != celoutils.ZeroAddress {
-			tokenIndexIter, err := chainProvider.NewBatchIterator(ctx, tokenIndex)
-			if err != nil {
-				lo.Error("could not create token index iter", "error", err)
-				os.Exit(1)
-			}
-
-			for {
-				batch, err := tokenIndexIter.Next(ctx)
-				if err != nil {
-					lo.Error("error fetching next token index batch", "error", err)
-					os.Exit(1)
-				}
-				if batch == nil {
-					break
-				}
-				lo.Debug("index batch", "index", tokenIndex.Hex(), "size", len(batch))
-				for _, address := range batch {
-					if address != celoutils.ZeroAddress {
-						var (
-							tokenName     string
-							tokenSymbol   string
-							tokenDecimals uint8
-						)
-
-						err := chainProvider.Client.CallCtx(
-							ctx,
-							eth.CallFunc(address, nameGetter).Returns(&tokenName),
-							eth.CallFunc(address, symbolGetter).Returns(&tokenSymbol),
-							eth.CallFunc(address, decimalsGetter).Returns(&tokenDecimals),
-						)
-						if err != nil {
-							lo.Error("error fetching token details", "error", err)
-							os.Exit(1)
-						}
-
-						if err := insertToken(ctx, TokenArgs{
-							ContractAddress: address.Hex(),
-							TokenName:       tokenName,
-							TokenSymbol:     tokenSymbol,
-							TokenDecimals:   tokenDecimals,
-						}); err != nil {
-							lo.Error("pg insert error", "error", err)
-							os.Exit(1)
-						}
-					}
-				}
-			}
-		}
-
-		if poolIndex := registryMap[celoutils.PoolIndex]; poolIndex != celoutils.ZeroAddress {
-			poolIndexIter, err := chainProvider.NewBatchIterator(ctx, poolIndex)
-			if err != nil {
-				lo.Error("cache could create pool index iter", "error", err)
-				os.Exit(1)
-			}
-
-			for {
-				batch, err := poolIndexIter.Next(ctx)
-				if err != nil {
-					lo.Error("error fetching next pool index batch", "error", err)
-					os.Exit(1)
-				}
-				if batch == nil {
-					break
-				}
-				lo.Debug("index batch", "index", poolIndex.Hex(), "size", len(batch))
-				for _, address := range batch {
-					var poolTokenIndex common.Address
-					err := chainProvider.Client.CallCtx(
-						ctx,
-						eth.CallFunc(address, tokenRegistryGetter).Returns(&poolTokenIndex),
-					)
-					if err != nil {
-						lo.Error("error fetching pool token index and/or quoter", "error", err)
-						os.Exit(1)
-					}
-					if poolTokenIndex != celoutils.ZeroAddress {
-						poolTokenIndexIter, err := chainProvider.NewBatchIterator(ctx, poolTokenIndex)
-						if err != nil {
-							lo.Error("error creating pool token index iter", "error", err)
-							os.Exit(1)
-						}
-
-						for {
-							batch, err := poolTokenIndexIter.Next(ctx)
-							if err != nil {
-								lo.Error("error fetching next pool token index batch", "error", err)
-								os.Exit(1)
-							}
-							if batch == nil {
-								break
-							}
-							lo.Debug("index batch", "index", poolTokenIndex.Hex(), "size", len(batch))
-							for _, address := range batch {
-								if address != celoutils.ZeroAddress {
-									var (
-										tokenName     string
-										tokenSymbol   string
-										tokenDecimals uint8
-									)
-
-									err := chainProvider.Client.CallCtx(
-										ctx,
-										eth.CallFunc(address, nameGetter).Returns(&tokenName),
-										eth.CallFunc(address, symbolGetter).Returns(&tokenSymbol),
-										eth.CallFunc(address, decimalsGetter).Returns(&tokenDecimals),
-									)
-									if err != nil {
-										lo.Error("error fetching token details", "error", err)
-										os.Exit(1)
-									}
-
-									if err := insertToken(ctx, TokenArgs{
-										ContractAddress: address.Hex(),
-										TokenName:       tokenName,
-										TokenSymbol:     tokenSymbol,
-										TokenDecimals:   tokenDecimals,
-									}); err != nil {
-										lo.Error("pg insert error", "error", err)
-										os.Exit(1)
-									}
-								}
-							}
-						}
-					}
-				}
-			}
-		}
-	}
-	lo.Info("tokens bootstrap complete")
-}
-
-func newPgStore() (*pgxpool.Pool, error) {
-	parsedConfig, err := pgxpool.ParseConfig(ko.MustString("postgres.dsn"))
-	if err != nil {
-		return nil, err
-	}
-
-	dbPool, err := pgxpool.NewWithConfig(context.Background(), parsedConfig)
-	if err != nil {
-		return nil, err
-	}
-
-	return dbPool, nil
-}
-
-func insertToken(ctx context.Context, insertArgs TokenArgs) error {
-	_, err := dbPool.Exec(
-		ctx,
-		insertTokenQuery,
-		insertArgs.ContractAddress,
-		insertArgs.TokenName,
-		insertArgs.TokenSymbol,
-		insertArgs.TokenDecimals,
-	)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}

cmd/init.go

@@ -1,4 +1,4 @@
-package util
+package main
 
 import (
 	"log/slog"
@@ -12,7 +12,7 @@ import (
 	"github.com/knadh/koanf/v2"
 )
 
-func InitLogger() *slog.Logger {
+func initLogger() *slog.Logger {
 	loggOpts := logg.LoggOpts{
 		FormatType: logg.Logfmt,
 		LogLevel:   slog.LevelInfo,
@@ -30,12 +30,12 @@ func InitLogger() *slog.Logger {
 	return logg.NewLogg(loggOpts)
 }
 
-func InitConfig(lo *slog.Logger, confFilePath string) *koanf.Koanf {
+func initConfig() *koanf.Koanf {
 	var (
 		ko = koanf.New(".")
 	)
 
-	confFile := file.Provider(confFilePath)
+	confFile := file.Provider(confFlag)
 	if err := ko.Load(confFile, toml.Parser()); err != nil {
 		lo.Error("could not parse configuration file", "error", err)
 		os.Exit(1)

cmd/main.go

@@ -5,18 +5,15 @@ import (
 	"errors"
 	"flag"
 	"log/slog"
-	"net/http"
 	"os"
 	"os/signal"
 	"sync"
 	"syscall"
 	"time"
 
-	"github.com/grassrootseconomics/celo-indexer/internal/api"
 	"github.com/grassrootseconomics/celo-indexer/internal/handler"
 	"github.com/grassrootseconomics/celo-indexer/internal/store"
 	"github.com/grassrootseconomics/celo-indexer/internal/sub"
-	"github.com/grassrootseconomics/celo-indexer/internal/util"
 	"github.com/knadh/koanf/v2"
 )
 
@@ -39,8 +36,8 @@ func init() {
 	flag.StringVar(&queriesFlag, "queries", "queries.sql", "Queries file location")
 	flag.Parse()
 
-	lo = util.InitLogger()
-	ko = util.InitConfig(lo, confFlag)
+	lo = initLogger()
+	ko = initConfig()
 
 	lo.Info("starting celo indexer", "build", build)
 }
@@ -49,6 +46,23 @@ func main() {
 	var wg sync.WaitGroup
 	ctx, stop := notifyShutdown()
 
+	// chain, err := chain.NewChainProvider(chain.ChainOpts{
+	// 	RPCEndpoint: ko.MustString("chain.rpc_endpoint"),
+	// 	ChainID:     ko.MustInt64("chain.chainid"),
+	// })
+	// if err != nil {
+	// 	lo.Error("chain provider bootstrap failed", "error", err)
+	// 	os.Exit(1)
+	// }
+	// cache := cache.NewCache()
+
+	// lo.Info("starting cache bootstrap this may take a few minutes")
+	// if err := chain.BootstrapCache(ko.MustStrings("bootstrap.ge_registries"), cache); err != nil {
+	// 	lo.Error("cache bootstrap failed", "error", err)
+	// 	os.Exit(1)
+	// }
+	// lo.Info("cache bootstrap completed successfully", "cache_size", cache.Size())
+
 	store, err := store.NewPgStore(store.PgOpts{
 		Logg:                 lo,
 		DSN:                  ko.MustString("postgres.dsn"),
@@ -62,6 +76,7 @@ func main() {
 
 	handler := handler.NewHandler(handler.HandlerOpts{
 		Store: store,
+		// Cache: cache,
 	})
 
 	jetStreamSub, err := sub.NewJetStreamSub(sub.JetStreamOpts{
@@ -76,27 +91,12 @@ func main() {
 		os.Exit(1)
 	}
 
-	apiServer := &http.Server{
-		Addr:    ko.MustString("api.address"),
-		Handler: api.New(),
-	}
-
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		jetStreamSub.Process()
 	}()
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		lo.Info("metrics and stats server starting", "address", ko.MustString("api.address"))
-		if err := apiServer.ListenAndServe(); err != http.ErrServerClosed {
-			lo.Error("failed to start API server", "error", err)
-			os.Exit(1)
-		}
-	}()
-
 	<-ctx.Done()
 	lo.Info("shutdown signal received")
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), defaultGracefulShutdownPeriod)
@@ -105,8 +105,6 @@ func main() {
 	go func() {
 		defer wg.Done()
 		jetStreamSub.Close()
-		store.Close()
-		apiServer.Shutdown(shutdownCtx)
 	}()
 
 	go func() {

config.toml

@@ -2,17 +2,15 @@
 go_process = true
 
 [api]
-address = ":5002"
+address = ":5001"
 
 [postgres]
 dsn = "postgres://postgres:postgres@127.0.0.1:5432/ge_celo_data"
 
-[jetstream]
-endpoint = "nats://127.0.0.1:4222"
-id = "celo-indexer-1"
-
 [chain]
-rpc_endpoint = "https://celo.grassecon.net"
+rpc_endpoint = "https://forno.celo.org"
+# Defaults to Celo mainnet
+# At the moment only support Celo based blockchains
 chainid = 42220
 
 [bootstrap]
@@ -22,3 +20,7 @@ ge_registries = [
     "0xd1FB944748aca327a1ba036B082993D9dd9Bfa0C",
     "0x0cc9f4fff962def35bb34a53691180b13e653030",
 ]
+
+[jetstream]
+endpoint = "nats://127.0.0.1:4222"
+id = "celo-indexer-1"

dev/docker-compose.postgres.yaml

@@ -1,22 +0,0 @@
-services:
-  postgres:
-    image: postgres:16-alpine
-    restart: unless-stopped
-    user: postgres
-    environment:
-      - POSTGRES_PASSWORD=postgres
-      - POSTGRES_USER=postgres
-    volumes:
-      - ./init_db.sql:/docker-entrypoint-initdb.d/init_db.sql
-      - indexer-pg:/var/lib/postgresql/data
-    ports:
-      - "127.0.0.1:5432:5432"
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready"]
-      interval: 10s
-      timeout: 5s
-      retries: 5
-
-volumes:
-  indexer-pg:
-    driver: local

dev/docker-compose.yaml

@@ -8,26 +8,15 @@ services:
       - POSTGRES_USER=postgres
     volumes:
       - ./init_db.sql:/docker-entrypoint-initdb.d/init_db.sql
-      - celo-indexer-pg:/var/lib/postgresql/data
+      - indexer-pg:/var/lib/postgresql/data
     ports:
       - "127.0.0.1:5432:5432"
     healthcheck:
       test: ["CMD-SHELL", "pg_isready"]
       interval: 10s
       timeout: 5s
-      retries: 5   
-  celo-indexer:
-    image: celo-indexer:latest
-    restart: unless-stopped
-    depends_on:
-      postgres:
-        condition: service_healthy
-    environment:
-    # Point to the correct celo-tracker JetStream instance
-      - INDEXER_JETSTREAM__ENDPOINT=nats://nats:4222
-    ports:
-      - 127.0.0.1:5002:5002
-      
+      retries: 5
+
 volumes:
-  celo-indexer-pg:
+  indexer-pg:
     driver: local

go.mod

@@ -3,11 +3,10 @@ module github.com/grassrootseconomics/celo-indexer
 go 1.22.3
 
 require (
-	github.com/VictoriaMetrics/metrics v1.33.1
 	github.com/celo-org/celo-blockchain v1.8.4
 	github.com/grassrootseconomics/celo-tracker v1.0.2-beta
-	github.com/grassrootseconomics/celoutils/v3 v3.3.1
-	github.com/grassrootseconomics/w3-celo v0.19.0
+	github.com/grassrootseconomics/celoutils/v3 v3.0.1
+	github.com/grassrootseconomics/w3-celo v0.17.2
 	github.com/jackc/pgx/v5 v5.6.0
 	github.com/jackc/tern/v2 v2.1.1
 	github.com/kamikazechaser/common v0.2.0
@@ -17,7 +16,7 @@ require (
 	github.com/knadh/koanf/providers/file v0.1.0
 	github.com/knadh/koanf/v2 v2.1.1
 	github.com/nats-io/nats.go v1.34.1
-	github.com/uptrace/bunrouter v1.0.21
+	github.com/puzpuzpuz/xsync/v3 v3.1.0
 )
 
 require (
@@ -77,8 +76,6 @@ require (
 	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
-	github.com/valyala/fastrand v1.1.0 // indirect
-	github.com/valyala/histogram v1.2.0 // indirect
 	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/net v0.23.0 // indirect
 	golang.org/x/sync v0.7.0 // indirect

go.sum

@@ -11,8 +11,6 @@ github.com/StackExchange/wmi v1.2.1 h1:VIkavFPXSjcnS+O8yTq7NI32k0R5Aj+v39y29VYDO
 github.com/StackExchange/wmi v1.2.1/go.mod h1:rcmrprowKIVzvc+NUiLncP2uuArMWLCbu9SBzvHz7e8=
 github.com/VictoriaMetrics/fastcache v1.12.1 h1:i0mICQuojGDL3KblA7wUNlY5lOK6a4bwt3uRKnkZU40=
 github.com/VictoriaMetrics/fastcache v1.12.1/go.mod h1:tX04vaqcNoQeGLD+ra5pU5sWkuxnzWhEzLwhP9w653o=
-github.com/VictoriaMetrics/metrics v1.33.1 h1:CNV3tfm2Kpv7Y9W3ohmvqgFWPR55tV2c7M2U6OIo+UM=
-github.com/VictoriaMetrics/metrics v1.33.1/go.mod h1:r7hveu6xMdUACXvB8TYdAj8WEsKzWB0EkpJN+RDtOf8=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8=
@@ -96,10 +94,10 @@ github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWm
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grassrootseconomics/celo-tracker v1.0.2-beta h1:iB5JC/GQDMQrml9A85myvxXp8HC2aGiiLO7VXRfbH0E=
 github.com/grassrootseconomics/celo-tracker v1.0.2-beta/go.mod h1:+byt6N9CSgaLgVfWfMxTKH9HguRFNW5uHv+jz668bFs=
-github.com/grassrootseconomics/celoutils/v3 v3.3.1 h1:ehf0eG2c8BfR2DojOm/nk8SQ44DyTFxXKISw1dHym/w=
-github.com/grassrootseconomics/celoutils/v3 v3.3.1/go.mod h1:HTHFNnWmvL6rY096xAraOMV3POYFAquoTABP6ogaF7U=
-github.com/grassrootseconomics/w3-celo v0.19.0 h1:3O5GK2os3/f9D/5oKsB0YxGyO5S3aqon/kx7ckyjQ4o=
-github.com/grassrootseconomics/w3-celo v0.19.0/go.mod h1:M3KJaj25DspF9siaqNXyamAuzC5DmWXx74tpyPp+R4Y=
+github.com/grassrootseconomics/celoutils/v3 v3.0.1 h1:R8W7jueJTFVhASV8ixDMaX5voJ2G/pW/tXLhf+KB43E=
+github.com/grassrootseconomics/celoutils/v3 v3.0.1/go.mod h1:VVo5Ge0Lop9rjVhTd0fnidyJmIolYAXsEJ6bzkangwQ=
+github.com/grassrootseconomics/w3-celo v0.17.2 h1:sEBlOe/H2F6YvBRTGD/kiObVqUFW7AczbzC8xxsVLDY=
+github.com/grassrootseconomics/w3-celo v0.17.2/go.mod h1:M3KJaj25DspF9siaqNXyamAuzC5DmWXx74tpyPp+R4Y=
 github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d h1:dg1dEPuWpEqDnvIw251EVy4zlP8gWbsGj4BsUKCRpYs=
 github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hdevalence/ed25519consensus v0.0.0-20201207055737-7fde80a9d5ff h1:LeVKjw8pcDQj7WVVnbFvbD7ovcv+r/l15ka1NH6Lswc=
@@ -197,6 +195,8 @@ github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7q
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/tsdb v0.7.1 h1:YZcsG11NqnK4czYLrWd9mpEuAJIHVQLwdrleYfszMAA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/puzpuzpuz/xsync/v3 v3.1.0 h1:EewKT7/LNac5SLiEblJeUu8z5eERHrmRLnMQL2d7qX4=
+github.com/puzpuzpuz/xsync/v3 v3.1.0/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.4.2 h1:YwD0ulJSJytLpiaWua0sBDusfsCZohxjxzVTYjwxfV8=
 github.com/rivo/uniseg v0.4.2/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
@@ -224,12 +224,6 @@ github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFA
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
 github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=
 github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY=
-github.com/uptrace/bunrouter v1.0.21 h1:HXarvX+N834sXyHpl+I/TuE11m19kLW/qG5u3YpHUag=
-github.com/uptrace/bunrouter v1.0.21/go.mod h1:TwT7Bc0ztF2Z2q/ZzMuSVkcb/Ig/d3MQeP2cxn3e1hI=
-github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8=
-github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
-github.com/valyala/histogram v1.2.0 h1:wyYGAZZt3CpwUiIb9AU/Zbllg1llXyrtApRS815OLoQ=
-github.com/valyala/histogram v1.2.0/go.mod h1:Hb4kBwb4UxsaNbbbh+RRz8ZR6pdodR57tzWUS3BUzXY=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=

internal/api/api.go

@@ -1,23 +0,0 @@
-package api
-
-import (
-	"net/http"
-
-	"github.com/VictoriaMetrics/metrics"
-	"github.com/uptrace/bunrouter"
-)
-
-func New() *bunrouter.Router {
-	router := bunrouter.New()
-
-	router.GET("/metrics", metricsHandler())
-
-	return router
-}
-
-func metricsHandler() bunrouter.HandlerFunc {
-	return func(w http.ResponseWriter, _ bunrouter.Request) error {
-		metrics.WritePrometheus(w, true)
-		return nil
-	}
-}

internal/cache/xmap.go

@@ -0,0 +1,37 @@
+package cache
+
+import (
+	"github.com/puzpuzpuz/xsync/v3"
+)
+
+type Cache struct {
+	xmap *xsync.Map
+}
+
+func NewCache() *Cache {
+	return &Cache{
+		xmap: xsync.NewMap(),
+	}
+}
+
+func (c *Cache) Purge() error {
+	c.xmap.Clear()
+	return nil
+}
+
+func (c *Cache) Exists(key string) bool {
+	_, ok := c.xmap.Load(key)
+	return ok
+}
+
+func (c *Cache) Add(key string) {
+	c.xmap.Store(key, nil)
+}
+
+func (c *Cache) Remove(key string) {
+	c.xmap.Delete(key)
+}
+
+func (c *Cache) Size() int {
+	return c.xmap.Size()
+}

internal/chain/chain.go

@@ -0,0 +1,121 @@
+package chain
+
+import (
+	"context"
+	"math/big"
+	"net/http"
+	"time"
+
+	"github.com/celo-org/celo-blockchain/common"
+	"github.com/celo-org/celo-blockchain/rpc"
+	"github.com/grassrootseconomics/celo-indexer/internal/cache"
+	"github.com/grassrootseconomics/celoutils/v3"
+	"github.com/grassrootseconomics/w3-celo"
+	"github.com/grassrootseconomics/w3-celo/module/eth"
+	"github.com/grassrootseconomics/w3-celo/w3types"
+)
+
+type (
+	ChainOpts struct {
+		RPCEndpoint string
+		ChainID     int64
+	}
+
+	Chain struct {
+		Provider *celoutils.Provider
+	}
+)
+
+const bootstrapTimeout = 2 * time.Minute
+
+var (
+	entryCountFunc = w3.MustNewFunc("entryCount()", "uint256")
+	entrySig       = w3.MustNewFunc("entry(uint256 _idx)", "address")
+)
+
+func NewChainProvider(o ChainOpts) (*Chain, error) {
+	customRPCClient, err := lowTimeoutRPCClient(o.RPCEndpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	chainProvider := celoutils.NewProvider(
+		o.RPCEndpoint,
+		o.ChainID,
+		celoutils.WithClient(customRPCClient),
+	)
+
+	return &Chain{
+		Provider: chainProvider,
+	}, nil
+}
+
+func lowTimeoutRPCClient(rpcEndpoint string) (*w3.Client, error) {
+	httpClient := &http.Client{
+		Timeout: 10 * time.Second,
+	}
+
+	rpcClient, err := rpc.DialHTTPWithClient(
+		rpcEndpoint,
+		httpClient,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return w3.NewClient(rpcClient), nil
+}
+
+func (c *Chain) BootstrapCache(registries []string, cache *cache.Cache) error {
+	var ZeroAddress common.Address
+	for _, registry := range registries {
+		ctx, cancel := context.WithTimeout(context.Background(), bootstrapTimeout)
+		defer cancel()
+
+		registryMap, err := c.Provider.RegistryMap(ctx, w3.A(registry))
+		if err != nil {
+			return err
+		}
+
+		if accountIndex := registryMap[celoutils.AccountIndex]; accountIndex != ZeroAddress {
+			if err := c.getAllAccountsFromAccountIndex(ctx, accountIndex, cache); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (c *Chain) getAllAccountsFromAccountIndex(ctx context.Context, accountIndex common.Address, cache *cache.Cache) error {
+	var accountIndexEntryCount big.Int
+
+	if err := c.Provider.Client.CallCtx(
+		ctx,
+		eth.CallFunc(accountIndex, entryCountFunc).Returns(&accountIndexEntryCount),
+	); err != nil {
+		return err
+	}
+
+	// TODO: Temporary skip custodial bootsrap, load from DB
+	if accountIndexEntryCount.Int64() > 1000 {
+		return nil
+	}
+
+	calls := make([]w3types.RPCCaller, accountIndexEntryCount.Int64())
+	accountAddresses := make([]common.Address, accountIndexEntryCount.Int64())
+
+	for i := 0; i < int(accountIndexEntryCount.Int64()); i++ {
+		calls[i] = eth.CallFunc(accountIndex, entrySig, new(big.Int).SetInt64(int64(i))).Returns(&accountAddresses[i])
+	}
+
+	if err := c.Provider.Client.CallCtx(ctx, calls...); err != nil {
+		return err
+	}
+
+	for _, address := range accountAddresses {
+		cache.Add(address.Hex())
+	}
+
+	return nil
+}

internal/handler/handler.go

@@ -11,16 +11,19 @@ import (
 type (
 	HandlerOpts struct {
 		Store store.Store
+		// Cache *cache.Cache
 	}
 
 	Handler struct {
 		store store.Store
+		// cache *cache.Cache
 	}
 )
 
 func NewHandler(o HandlerOpts) *Handler {
 	return &Handler{
 		store: o.Store,
+		// cache: o.Cache,
 	}
 }
 
@@ -33,6 +36,12 @@ func (h *Handler) Handle(ctx context.Context, msgSubject string, msgData []byte)
 
 	switch msgSubject {
 	case "TRACKER.TOKEN_TRANSFER":
+		// from := chainEvent.Payload["from"].(string)
+		// to := chainEvent.Payload["to"].(string)
+
+		// if h.cache.Exists(from) || h.cache.Exists(to) {
+		// 	return h.store.InsertTokenTransfer(ctx, chainEvent)
+		// }
 		return h.store.InsertTokenTransfer(ctx, chainEvent)
 	case "TRACKER.POOL_SWAP":
 		return h.store.InsertPoolSwap(ctx, chainEvent)

internal/store/pg.go

@@ -37,6 +37,7 @@ type (
 		InsertPoolSwap         string `query:"insert-pool-swap"`
 		InsertPoolDeposit      string `query:"insert-pool-deposit"`
 		InsertPriceQuoteUpdate string `query:"insert-price-quote-update"`
+		CheckAddressExists     string `query:"address-exists"`
 	}
 )
 
@@ -68,10 +69,6 @@ func NewPgStore(o PgOpts) (Store, error) {
 	}, nil
 }
 
-func (pg *Pg) Close() {
-	pg.db.Close()
-}
-
 func (pg *Pg) InsertTokenTransfer(ctx context.Context, eventPayload event.Event) error {
 	return pg.executeTransaction(ctx, func(tx pgx.Tx) error {
 		txID, err := pg.insertTx(ctx, tx, eventPayload)
@@ -86,7 +83,6 @@ func (pg *Pg) InsertTokenTransfer(ctx context.Context, eventPayload event.Event)
 			eventPayload.Payload["from"].(string),
 			eventPayload.Payload["to"].(string),
 			eventPayload.Payload["value"].(string),
-			eventPayload.ContractAddress,
 		)
 		return err
 	})
@@ -106,7 +102,6 @@ func (pg *Pg) InsertTokenMint(ctx context.Context, eventPayload event.Event) err
 			eventPayload.Payload["tokenMinter"].(string),
 			eventPayload.Payload["to"].(string),
 			eventPayload.Payload["value"].(string),
-			eventPayload.ContractAddress,
 		)
 		return err
 	})
@@ -125,7 +120,6 @@ func (pg *Pg) InsertTokenBurn(ctx context.Context, eventPayload event.Event) err
 			txID,
 			eventPayload.Payload["tokenBurner"].(string),
 			eventPayload.Payload["value"].(string),
-			eventPayload.ContractAddress,
 		)
 		return err
 	})
@@ -145,7 +139,6 @@ func (pg *Pg) InsertFaucetGive(ctx context.Context, eventPayload event.Event) er
 			eventPayload.Payload["token"].(string),
 			eventPayload.Payload["recipient"].(string),
 			eventPayload.Payload["amount"].(string),
-			eventPayload.ContractAddress,
 		)
 		return err
 	})
@@ -168,7 +161,6 @@ func (pg *Pg) InsertPoolSwap(ctx context.Context, eventPayload event.Event) erro
 			eventPayload.Payload["amountIn"].(string),
 			eventPayload.Payload["amountOut"].(string),
 			eventPayload.Payload["fee"].(string),
-			eventPayload.ContractAddress,
 		)
 		return err
 	})
@@ -188,7 +180,6 @@ func (pg *Pg) InsertPoolDeposit(ctx context.Context, eventPayload event.Event) e
 			eventPayload.Payload["initiator"].(string),
 			eventPayload.Payload["tokenIn"].(string),
 			eventPayload.Payload["amountIn"].(string),
-			eventPayload.ContractAddress,
 		)
 		return err
 	})
@@ -207,7 +198,6 @@ func (pg *Pg) InsertPriceQuoteUpdate(ctx context.Context, eventPayload event.Eve
 			txID,
 			eventPayload.Payload["token"].(string),
 			eventPayload.Payload["exchangeRate"].(string),
-			eventPayload.ContractAddress,
 		)
 		return err
 	})
@@ -220,6 +210,7 @@ func (pg *Pg) insertTx(ctx context.Context, tx pgx.Tx, eventPayload event.Event)
 		pg.queries.InsertTx,
 		eventPayload.TxHash,
 		eventPayload.Block,
+		eventPayload.ContractAddress,
 		time.Unix(int64(eventPayload.Timestamp), 0).UTC(),
 		eventPayload.Success,
 	).Scan(&txID); err != nil {

internal/store/store.go

@@ -15,6 +15,5 @@ type (
 		InsertPoolSwap(context.Context, event.Event) error
 		InsertPoolDeposit(context.Context, event.Event) error
 		InsertPriceQuoteUpdate(context.Context, event.Event) error
-		Close()
 	}
 )

migrations/001_indexer_base.sql

@@ -65,6 +65,7 @@ CREATE TABLE IF NOT EXISTS price_index_updates (
   exchange_rate NUMERIC NOT NULL
 );
 
+
 CREATE TABLE IF NOT EXISTS contracts (
   id INT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
   contract_address VARCHAR(42) UNIQUE NOT NULL,

migrations/002_fix_contract_address.sql

@@ -1,29 +0,0 @@
-ALTER TABLE tx DROP COLUMN contract_address;
-
-ALTER TABLE token_transfer ADD COLUMN contract_address VARCHAR(42);
-UPDATE token_transfer SET contract_address = '0x0000000000000000000000000000000000000000';
-ALTER TABLE token_transfer ALTER COLUMN contract_address SET NOT NULL;
-
-ALTER TABLE token_mint ADD COLUMN contract_address VARCHAR(42);
-UPDATE token_mint SET contract_address = '0x0000000000000000000000000000000000000000';
-ALTER TABLE token_mint ALTER COLUMN contract_address SET NOT NULL;
-
-ALTER TABLE token_burn ADD COLUMN contract_address VARCHAR(42);
-UPDATE token_burn SET contract_address = '0x0000000000000000000000000000000000000000';
-ALTER TABLE token_burn ALTER COLUMN contract_address SET NOT NULL;
-
-ALTER TABLE faucet_give ADD COLUMN contract_address VARCHAR(42);
-UPDATE faucet_give SET contract_address = '0x0000000000000000000000000000000000000000';
-ALTER TABLE faucet_give ALTER COLUMN contract_address SET NOT NULL;
-
-ALTER TABLE pool_swap ADD COLUMN contract_address VARCHAR(42);
-UPDATE pool_swap SET contract_address = '0x0000000000000000000000000000000000000000';
-ALTER TABLE pool_swap ALTER COLUMN contract_address SET NOT NULL;
-
-ALTER TABLE pool_deposit ADD COLUMN contract_address VARCHAR(42);
-UPDATE pool_deposit SET contract_address = '0x0000000000000000000000000000000000000000';
-ALTER TABLE pool_deposit ALTER COLUMN contract_address SET NOT NULL;
-
-ALTER TABLE price_index_updates ADD COLUMN contract_address VARCHAR(42);
-UPDATE price_index_updates SET contract_address = '0x0000000000000000000000000000000000000000';
-ALTER TABLE price_index_updates ALTER COLUMN contract_address SET NOT NULL;

migrations/003_fix_tokens_constraint.sql

@@ -1 +0,0 @@
-ALTER TABLE tokens ALTER token_version DROP NOT NULL, ALTER token_type DROP NOT NULL;

queries.sql

@@ -1,15 +1,17 @@
 --name: insert-tx
 -- $1: tx_hash
 -- $2: block_number
--- $3: date_block
--- $4: success
+-- $3: contract_address
+-- $4: date_block
+-- $5: success
 WITH insert_tx AS (
     INSERT INTO tx(
         tx_hash,
         block_number,
+        contract_address,
         date_block,
         success
-    ) VALUES($1, $2, $3, $4) ON CONFLICT DO NOTHING RETURNING id
+    ) VALUES($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING RETURNING id
 )
 SELECT id FROM insert_tx
 UNION ALL
@@ -21,54 +23,46 @@ LIMIT 1
 -- $2: sender_address
 -- $3: recipient_address
 -- $4: transfer_value
--- $5: contract_address
 INSERT INTO token_transfer(
     tx_id,
     sender_address,
     recipient_address,
-    transfer_value,
-    contract_address
-) VALUES($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING
+    transfer_value
+) VALUES($1, $2, $3, $4) ON CONFLICT DO NOTHING
 
 --name: insert-token-mint
 -- $1: tx_id
 -- $2: minter_address
 -- $3: recipient_address
 -- $4: mint_value
--- $5: contract_address
 INSERT INTO token_mint(
     tx_id,
     minter_address,
     recipient_address,
-    mint_value,
-    contract_address
-) VALUES($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING
+    mint_value
+) VALUES($1, $2, $3, $4) ON CONFLICT DO NOTHING
 
 --name: insert-token-burn
 -- $1: tx_id
 -- $2: burner_address
 -- $3: burn_value
--- $4: contract_address
 INSERT INTO token_burn(
     tx_id,
     burner_address,
-    burn_value,
-    contract_address
-) VALUES($1, $2, $3, $4) ON CONFLICT DO NOTHING
+    burn_value
+) VALUES($1, $2, $3) ON CONFLICT DO NOTHING
 
 --name: insert-faucet-give
 -- $1: tx_id
 -- $2: token_address
 -- $3: recipient_address
 -- $4: give_value
--- $5: contract_address
 INSERT INTO faucet_give(
     tx_id,
     token_address,
     recipient_address,
-    give_value,
-    contract_address
-) VALUES($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING
+    give_value
+) VALUES($1, $2, $3, $4) ON CONFLICT DO NOTHING
 
 --name: insert-pool-swap
 -- $1: tx_id
@@ -78,7 +72,6 @@ INSERT INTO faucet_give(
 -- $5: in_value
 -- $6: out_value
 -- $7: fee
--- $8: contract_address
 INSERT INTO pool_swap(
     tx_id,
     initiator_address,
@@ -86,32 +79,37 @@ INSERT INTO pool_swap(
     token_out_address,
     in_value,
     out_value,
-    fee,
-    contract_address
-) VALUES($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT DO NOTHING
+    fee
+) VALUES($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING
 
 --name: insert-pool-deposit
 -- $1: tx_id
 -- $2: initiator_address
 -- $3: token_in_address
--- $4: in_value
--- $5: contract_address
+-- $4: token_out_address
+-- $5: in_value
+-- $6: out_value
+-- $7: fee
 INSERT INTO pool_deposit(
     tx_id,
     initiator_address,
     token_in_address,
-    in_value,
-    contract_address
-) VALUES($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING
+    in_value
+) VALUES($1, $2, $3, $4) ON CONFLICT DO NOTHING
 
 --name: insert-price-quote-update
 -- $1: tx_id
 -- $2: token
 -- $3: exchange_rate
--- $4: contract_address
 INSERT INTO price_index_updates(
     tx_id,
     token,
-    exchange_rate,
-    contract_address
-) VALUES($1, $2, $3, $4) ON CONFLICT DO NOTHING
+    exchange_rate
+) VALUES($1, $2, $3) ON CONFLICT DO NOTHING
+
+--name: address-exists
+-- $1: blockchain_address_1
+-- $2: blockchain_address_2
+SELECT EXISTS(
+    SELECT 1 FROM sarafu_network.accounts WHERE blockchain_address=$1 OR blockchain_address=$2
+)