diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..0720f68 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +.github +.deepsource.toml +.goreleaser.yaml + +**/.env +**/.git +**/.gitignore + +**/docker-compose* +**/Dockerfile* \ No newline at end of file diff --git a/.env.example b/.env.example index 41166c6..90d70fc 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,5 @@ -export PG_HOST=localhost -export PG_PORT=5432 -export PG_DB=cic_indexer -export PG_USER=postgres -export PG_PASSWORD=postgres +EVENTS_CHAIN__GRAPHQL_ENDPOINT= +EVENTS_CHAIN__WS_ENDPOINT= +EVENTS_SYNCER__INITIAL_LOWER_BOUND= +EVENTS_POSTGRES__DSN= +EVENTS_JETSTREAM__ENDPOINT= diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml deleted file mode 100644 index ee2032b..0000000 --- a/.github/workflows/build.yaml +++ /dev/null @@ -1,58 +0,0 @@ -name: build - -on: - push: - tags: - - "v*" - -jobs: - goreleaser: - runs-on: ubuntu-latest - container: - image: goreleaser/goreleaser-cross - environment: build - steps: - - name: Checkout - uses: actions/checkout@v3 - with: - fetch-depth: 0 - - - name: Workaround Git Security Warning - run: | - # Workaround a bug in github actions: - # https://github.com/actions/runner-images/issues/6775. - git config --global --add safe.directory "$GITHUB_WORKSPACE" - - - name: Set up Go - uses: actions/setup-go@v2 - with: - go-version: 1.20.1 - - - name: Run GoReleaser - uses: goreleaser/goreleaser-action@v2 - with: - version: latest - args: release --rm-dist - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - build-publish: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v2 - - uses: docker/login-action@v1.14.1 - with: - registry: ghcr.io - username: ${{ github.repository_owner }} - password: ${{ secrets.GITHUB_TOKEN }} - - name: docker compose build push - env: - DOCKER_BUILDKIT: 1 - COMPOSE_DOCKER_CLI_BUILD: 1 - run: | - export TAG=$GITHUB_REF_NAME - docker-compose -f docker-compose.build.yaml build --progress plain - docker-compose -f docker-compose.build.yaml push - export TAG=latest - docker-compose -f docker-compose.build.yaml build --progress plain - docker-compose -f docker-compose.build.yaml push - \ No newline at end of file diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml new file mode 100644 index 0000000..4eae179 --- /dev/null +++ b/.github/workflows/release.yaml @@ -0,0 +1,100 @@ +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_COMMIT=${{ env.RELEASE_SHORT_COMMIT }} + cache-from: type=local,src=/tmp/.buildx-cache + cache-to: type=local,dest=/tmp/.buildx-cache + tags: | + ghcr.io/grassrootseconomics/cic-chain-events/cic-chain-events:latest + ghcr.io/grassrootseconomics/cic-chain-events/cic-chain-events:${{ env.RELEASE_TAG }} + + goreleaser: + runs-on: ubuntu-latest + container: + image: goreleaser/goreleaser-cross + environment: build + + steps: + + - name: Checkout + uses: actions/checkout@v3 + with: + fetch-depth: 0 + + - uses: actions/cache@v3 + with: + path: | + ~/go/pkg/mod + ~/.cache/go-build + ~/Library/Caches/go-build + %LocalAppData%\go-build + key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }} + restore-keys: | + ${{ runner.os }}-go- + + - name: Workaround Git Security Warning + run: | + # Workaround a bug in github actions: + # https://github.com/actions/runner-images/issues/6775. + git config --global --add safe.directory "$GITHUB_WORKSPACE" + + - name: Set up Go + uses: actions/setup-go@v3 + with: + go-version: 'stable' + + - name: Run GoReleaser + uses: goreleaser/goreleaser-action@v4 + with: + version: latest + args: release --rm-dist + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.goreleaser.yaml b/.goreleaser.yaml index 68b369f..dbb022d 100644 --- a/.goreleaser.yaml +++ b/.goreleaser.yaml @@ -8,9 +8,9 @@ builds: - linux goarch: - amd64 - main: ./cmd + main: ./cmd/service ldflags: - - -s -w + - -X main.build={{.ShortCommit}} -s -w archives: - format: tar.gz diff --git a/Dockerfile b/Dockerfile index 5eb1998..8b6999d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,25 +1,33 @@ -FROM golang:1.19-bullseye as build +FROM golang:1-bullseye as build + +ENV CGO_ENABLED=1 +ENV GOOS=linux +ENV GOARCH=amd64 + +ENV BUILD_COMMIT=${BUILD_COMMIT} WORKDIR /build + +COPY go.* . +RUN go mod download + COPY . . -RUN CGO_ENABLED=1 GOOS=linux GOARCH=amd64 go build -o cic-chain-events -ldflags="-s -w" cmd/*.go +RUN go build -o cic-chain-events -ldflags="-X main.build=${BUILD_COMMIT} -s -w" cmd/service/* + FROM debian:bullseye-slim ENV DEBIAN_FRONTEND=noninteractive -RUN set -x && \ - apt-get update && \ - apt-get install -y ca-certificates && \ - rm -rf /var/cache/apt/archives /var/lib/apt/lists/* WORKDIR /service +COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ COPY --from=build /build/cic-chain-events . +COPY migrations migrations/ COPY config.toml . COPY queries.sql . COPY LICENSE . -COPY migrations migrations/ + +EXPOSE 5000 CMD ["./cic-chain-events"] - -EXPOSE 8080 \ No newline at end of file diff --git a/Makefile b/Makefile index e6edbf0..f5fdaf1 100644 --- a/Makefile +++ b/Makefile @@ -1,27 +1,17 @@ -SHELL := /bin/bash BIN := cic-chain-events +BUILD_CONF := CGO_ENABLED=1 GOOS=linux GOARCH=amd64 +BUILD_COMMIT := $(shell git rev-parse --short HEAD 2> /dev/null) .PHONY: build +clean: + rm ${BIN} + build: - CGO_ENABLED=1 go build -v -ldflags="-s -w" -o ${BIN} cmd/*.go + ${BUILD_CONF} go build -ldflags="-X main.build=${BUILD_COMMIT} -s -w" -o ${BIN} cmd/service/*.go run: - CGO_ENABLED=1 go run -ldflags="-s -w" cmd/*.go + ${BUILD_CONF} go run cmd/service/* -mod: - go mod tidy - go mod verify - -test-pkg: - TEST_GRAPHQL_ENDPOINT=https://rpc.alfajores.celo.grassecon.net/graphql go test -v -covermode atomic -coverprofile=covprofile ./pkg/... - -bench-pkg: - TEST_RPC_ENDPOINT=https://rpc.alfajores.celo.grassecon.net TEST_GRAPHQL_ENDPOINT=https://rpc.alfajores.celo.grassecon.net/graphql go test -v -bench=. -run=^Benchmark ./pkg/... - -migrate: - tern migrate -c migrations/tern.conf - -docker-clean: - docker-compose down - docker volume rm cic-chain-events_cic-indexer-pg cic-chain-events_cic-indexer-nats +run-debug: + ${BUILD_CONF} go run cmd/service/* -debug diff --git a/cmd/init.go b/cmd/init.go deleted file mode 100644 index e085194..0000000 --- a/cmd/init.go +++ /dev/null @@ -1,93 +0,0 @@ -package main - -import ( - "strings" - "time" - - "github.com/grassrootseconomics/cic-chain-events/internal/events" - "github.com/grassrootseconomics/cic-chain-events/internal/store" - "github.com/grassrootseconomics/cic-chain-events/pkg/fetch" - "github.com/jackc/pgx/v5" - "github.com/knadh/goyesql/v2" - "github.com/knadh/koanf" - "github.com/knadh/koanf/parsers/toml" - "github.com/knadh/koanf/providers/env" - "github.com/knadh/koanf/providers/file" - "github.com/zerodha/logf" -) - -func initLogger(debug bool) logf.Logger { - loggOpts := logf.Opts{ - EnableColor: true, - } - - if debug { - loggOpts.Level = logf.DebugLevel - loggOpts.EnableCaller = true - } - - return logf.New(loggOpts) -} - -func initConfig(configFilePath string) *koanf.Koanf { - var ( - ko = koanf.New(".") - ) - - confFile := file.Provider(configFilePath) - if err := ko.Load(confFile, toml.Parser()); err != nil { - lo.Fatal("could not load config file", "error", err) - } - - if err := ko.Load(env.Provider("", ".", func(s string) string { - return strings.ReplaceAll(strings.ToLower( - strings.TrimPrefix(s, "")), "_", ".") - }), nil); err != nil { - lo.Fatal("could not override config from env vars", "error", err) - } - - return ko -} - -func initQueries(queriesPath string) goyesql.Queries { - queries, err := goyesql.ParseFile(queriesPath) - if err != nil { - lo.Fatal("could not load queries file", "error", err) - } - - return queries -} - -func initPgStore() (store.Store[pgx.Rows], error) { - pgStore, err := store.NewPostgresStore(store.PostgresStoreOpts{ - DSN: ko.MustString("postgres.dsn"), - InitialLowerBound: uint64(ko.MustInt64("syncer.initial_lower_bound")), - Logg: lo, - Queries: q, - }) - if err != nil { - return nil, err - } - - return pgStore, nil -} - -func initFetcher() fetch.Fetch { - return fetch.NewGraphqlFetcher(fetch.GraphqlOpts{ - GraphqlEndpoint: ko.MustString("chain.graphql_endpoint"), - }) -} - -func initJetStream() (events.EventEmitter, error) { - jsEmitter, err := events.NewJetStreamEventEmitter(events.JetStreamOpts{ - ServerUrl: ko.MustString("jetstream.endpoint"), - PersistDuration: time.Duration(ko.MustInt("jetstream.persist_duration_hours")) * time.Hour, - DedupDuration: time.Duration(ko.MustInt("jetstream.dedup_duration_hours")) * time.Hour, - }) - - if err != nil { - return nil, err - } - - return jsEmitter, nil -} diff --git a/cmd/api.go b/cmd/service/api.go similarity index 100% rename from cmd/api.go rename to cmd/service/api.go diff --git a/cmd/filters.go b/cmd/service/filters.go similarity index 71% rename from cmd/filters.go rename to cmd/service/filters.go index 8656835..b26bd27 100644 --- a/cmd/filters.go +++ b/cmd/service/filters.go @@ -4,8 +4,8 @@ import ( "strings" "sync" - "github.com/grassrootseconomics/cic-chain-events/internal/events" "github.com/grassrootseconomics/cic-chain-events/internal/filter" + "github.com/grassrootseconomics/cic-chain-events/internal/pub" ) var ( @@ -29,25 +29,25 @@ func initAddressFilter() filter.Filter { }) } -func initTransferFilter(eventEmitter events.EventEmitter) filter.Filter { +func initTransferFilter(pub *pub.Pub) filter.Filter { return filter.NewTransferFilter(filter.TransferFilterOpts{ - EventEmitter: eventEmitter, - Logg: lo, + Pub: pub, + Logg: lo, }) } -func initGasGiftFilter(eventEmitter events.EventEmitter) filter.Filter { +func initGasGiftFilter(pub *pub.Pub) filter.Filter { return filter.NewGasFilter(filter.GasFilterOpts{ - EventEmitter: eventEmitter, + Pub: pub, Logg: lo, SystemAddress: systemAddress, }) } -func initRegisterFilter(eventEmitter events.EventEmitter) filter.Filter { +func initRegisterFilter(pub *pub.Pub) filter.Filter { return filter.NewRegisterFilter(filter.RegisterFilterOpts{ - EventEmitter: eventEmitter, - Logg: lo, + Pub: pub, + Logg: lo, }) } diff --git a/cmd/service/init.go b/cmd/service/init.go new file mode 100644 index 0000000..f5798e7 --- /dev/null +++ b/cmd/service/init.go @@ -0,0 +1,130 @@ +package main + +import ( + "context" + "strings" + "time" + + "github.com/alitto/pond" + "github.com/grassrootseconomics/cic-chain-events/internal/pool" + "github.com/grassrootseconomics/cic-chain-events/internal/pub" + "github.com/grassrootseconomics/cic-chain-events/internal/store" + "github.com/grassrootseconomics/cic-chain-events/pkg/fetch" + "github.com/jackc/pgx/v5" + "github.com/knadh/goyesql/v2" + "github.com/knadh/koanf/parsers/toml" + "github.com/knadh/koanf/providers/env" + "github.com/knadh/koanf/providers/file" + "github.com/knadh/koanf/v2" + "github.com/nats-io/nats.go" + "github.com/zerodha/logf" +) + +func initLogger() logf.Logger { + loggOpts := logf.Opts{} + + if debugFlag { + loggOpts.EnableColor = true + loggOpts.EnableColor = true + loggOpts.Level = logf.DebugLevel + } + + return logf.New(loggOpts) +} + +func initConfig() *koanf.Koanf { + var ( + ko = koanf.New(".") + ) + + confFile := file.Provider(confFlag) + if err := ko.Load(confFile, toml.Parser()); err != nil { + lo.Fatal("init: could not load config file", "error", err) + } + + if err := ko.Load(env.Provider("EVENTS_", ".", func(s string) string { + return strings.ReplaceAll(strings.ToLower( + strings.TrimPrefix(s, "EVENTS_")), "__", ".") + }), nil); err != nil { + lo.Fatal("init: could not override config from env vars", "error", err) + } + + if debugFlag { + ko.Print() + } + + return ko +} + +func initQueries(queriesPath string) goyesql.Queries { + queries, err := goyesql.ParseFile(queriesPath) + if err != nil { + lo.Fatal("init: could not load queries file", "error", err) + } + + return queries +} + +func initPgStore(migrationsPath string, queries goyesql.Queries) store.Store[pgx.Rows] { + pgStore, err := store.NewPostgresStore(store.PostgresStoreOpts{ + MigrationsFolderPath: migrationsPath, + DSN: ko.MustString("postgres.dsn"), + InitialLowerBound: uint64(ko.MustInt64("syncer.initial_lower_bound")), + Logg: lo, + Queries: queries, + }) + if err != nil { + lo.Fatal("init: critical error loading chain provider", "error", err) + } + + return pgStore +} + +func initFetcher() fetch.Fetch { + return fetch.NewGraphqlFetcher(fetch.GraphqlOpts{ + GraphqlEndpoint: ko.MustString("chain.graphql_endpoint"), + }) +} + +func initJanitorWorkerPool(ctx context.Context) *pond.WorkerPool { + return pool.NewPool(ctx, pool.Opts{ + Concurrency: ko.MustInt("syncer.janitor_concurrency"), + QueueSize: ko.MustInt("syncer.janitor_queue_size"), + }) +} + +func initHeadSyncerWorkerPool(ctx context.Context) *pond.WorkerPool { + return pool.NewPool(ctx, pool.Opts{ + Concurrency: 1, + QueueSize: 1, + }) +} + +func initJetStream() (*nats.Conn, nats.JetStreamContext) { + natsConn, err := nats.Connect(ko.MustString("jetstream.endpoint")) + if err != nil { + lo.Fatal("init: critical error connecting to NATS", "error", err) + } + + js, err := natsConn.JetStream() + if err != nil { + lo.Fatal("init: bad JetStream opts", "error", err) + + } + + return natsConn, js +} + +func initPub(natsConn *nats.Conn, jsCtx nats.JetStreamContext) *pub.Pub { + pub, err := pub.NewPub(pub.PubOpts{ + DedupDuration: time.Duration(ko.MustInt("jetstream.dedup_duration_hrs")) * time.Hour, + JsCtx: jsCtx, + NatsConn: natsConn, + PersistDuration: time.Duration(ko.MustInt("jetstream.persist_duration_hrs")) * time.Hour, + }) + if err != nil { + lo.Fatal("init: critical error bootstrapping pub", "error", err) + } + + return pub +} diff --git a/cmd/main.go b/cmd/service/main.go similarity index 57% rename from cmd/main.go rename to cmd/service/main.go index e8e31bb..d6f3929 100644 --- a/cmd/main.go +++ b/cmd/service/main.go @@ -3,26 +3,34 @@ package main import ( "context" "flag" - "os" - "os/signal" "strings" "sync" - "syscall" "time" "github.com/grassrootseconomics/cic-chain-events/internal/filter" "github.com/grassrootseconomics/cic-chain-events/internal/pipeline" - "github.com/grassrootseconomics/cic-chain-events/internal/pool" + "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" + "github.com/knadh/koanf/v2" + "github.com/labstack/echo/v4" "github.com/zerodha/logf" ) +type ( + internalServiceContainer struct { + apiService *echo.Echo + pub *pub.Pub + } +) + var ( - confFlag string - debugFlag bool - queriesFlag string + build string + + confFlag string + debugFlag bool + migrationsFolderFlag string + queriesFlag string ko *koanf.Koanf lo logf.Logger @@ -31,61 +39,51 @@ var ( func init() { flag.StringVar(&confFlag, "config", "config.toml", "Config file location") - flag.BoolVar(&debugFlag, "log", true, "Enable debug logging") + flag.BoolVar(&debugFlag, "debug", false, "Enable debug logging") + flag.StringVar(&migrationsFolderFlag, "migrations", "migrations/", "Migrations folder location") flag.StringVar(&queriesFlag, "queries", "queries.sql", "Queries file location") flag.Parse() - lo = initLogger(debugFlag) - ko = initConfig(confFlag) - q = initQueries(queriesFlag) + lo = initLogger() + ko = initConfig() } func main() { - syncerStats := &syncer.Stats{} - wg := &sync.WaitGroup{} - apiServer := initApiServer() - - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer stop() - - janitorWorkerPool := pool.NewPool(ctx, pool.Opts{ - Concurrency: ko.MustInt("syncer.janitor_concurrency"), - QueueSize: ko.MustInt("syncer.janitor_queue_size"), - }) - - pgStore, err := initPgStore() - if err != nil { - lo.Fatal("main: critical error loading pg store", "error", err) - } - - jsCtx, err := initJetStream() - if err != nil { - lo.Fatal("main: critical error loading jetstream context", "error", err) - } + lo.Info("main: starting cic-chain-events", "build", build) + parsedQueries := initQueries(queriesFlag) graphqlFetcher := initFetcher() + pgStore := initPgStore(migrationsFolderFlag, parsedQueries) + natsConn, jsCtx := initJetStream() + jsPub := initPub(natsConn, jsCtx) pipeline := pipeline.NewPipeline(pipeline.PipelineOpts{ BlockFetcher: graphqlFetcher, Filters: []filter.Filter{ initAddressFilter(), - initGasGiftFilter(jsCtx), - initTransferFilter(jsCtx), - initRegisterFilter(jsCtx), + initGasGiftFilter(jsPub), + initTransferFilter(jsPub), + initRegisterFilter(jsPub), }, Logg: lo, Store: pgStore, }) - headSyncerWorker := pool.NewPool(ctx, pool.Opts{ - Concurrency: 1, - QueueSize: 1, - }) + internalServices := &internalServiceContainer{ + pub: jsPub, + } + syncerStats := &syncer.Stats{} + wg := &sync.WaitGroup{} + + signalCh, closeCh := createSigChannel() + defer closeCh() + + ctx, cancel := context.WithCancel(context.Background()) headSyncer, err := syncer.NewHeadSyncer(syncer.HeadSyncerOpts{ Logg: lo, Pipeline: pipeline, - Pool: headSyncerWorker, + Pool: initHeadSyncerWorkerPool(ctx), Stats: syncerStats, WsEndpoint: ko.MustString("chain.ws_endpoint"), }) @@ -97,7 +95,7 @@ func main() { BatchSize: uint64(ko.MustInt64("syncer.janitor_queue_size")), Logg: lo, Pipeline: pipeline, - Pool: janitorWorkerPool, + Pool: initJanitorWorkerPool(ctx), Stats: syncerStats, Store: pgStore, SweepInterval: time.Second * time.Duration(ko.MustInt64("syncer.janitor_sweep_interval")), @@ -107,6 +105,7 @@ func main() { go func() { defer wg.Done() if err := headSyncer.Start(ctx); err != nil { + lo.Info("main: starting head syncer") lo.Fatal("main: critical error starting head syncer", "error", err) } }() @@ -114,16 +113,19 @@ func main() { wg.Add(1) go func() { defer wg.Done() + lo.Info("main: starting janitor") if err := janitor.Start(ctx); err != nil { lo.Fatal("main: critical error starting janitor", "error", err) } }() + internalServices.apiService = initApiServer() wg.Add(1) go func() { defer wg.Done() - lo.Info("starting API server") - if err := apiServer.Start(ko.MustString("api.address")); err != nil { + host := ko.MustString("service.address") + lo.Info("main: starting API server", "host", host) + if err := internalServices.apiService.Start(host); err != nil { if strings.Contains(err.Error(), "Server closed") { lo.Info("main: shutting down server") } else { @@ -132,11 +134,9 @@ func main() { } }() - <-ctx.Done() - - if err := apiServer.Shutdown(ctx); err != nil { - lo.Error("main: could not gracefully shutdown api server", "err", err) - } + lo.Info("main: graceful shutdown triggered", "signal", <-signalCh) + cancel() + startGracefulShutdown(context.Background(), internalServices) wg.Wait() } diff --git a/cmd/service/utils.go b/cmd/service/utils.go new file mode 100644 index 0000000..0a89073 --- /dev/null +++ b/cmd/service/utils.go @@ -0,0 +1,29 @@ +package main + +import ( + "context" + "os" + "os/signal" + "syscall" + "time" +) + +func createSigChannel() (chan os.Signal, func()) { + signalCh := make(chan os.Signal, 1) + signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM, syscall.SIGINT) + + return signalCh, func() { + close(signalCh) + } +} + +func startGracefulShutdown(ctx context.Context, internalServices *internalServiceContainer) { + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + internalServices.pub.Close() + + if err := internalServices.apiService.Shutdown(ctx); err != nil { + lo.Fatal("Could not gracefully shutdown api server", "err", err) + } +} diff --git a/config.toml b/config.toml index dbfa45b..7c606fe 100644 --- a/config.toml +++ b/config.toml @@ -3,33 +3,33 @@ go_process = true # API server -[api] +[service] # Host and port -address = ":8085" +address = ":5000" # Geth API endpoints [chain] -graphql_endpoint = "https://rpc.alfajores.celo.grassecon.net/graphql" -ws_endpoint = "wss://ws.alfajores.celo.grassecon.net" +graphql_endpoint = "" +ws_endpoint = "" # Syncer configs [syncer] # Maximum number of missing blocks pushed into the worker queue every janitor sweep -janitor_queue_size = 500 +janitor_queue_size = 250 # Number of goroutines assigned to the worker pool -janitor_concurrency = 5 +janitor_concurrency = 3 # Syncer start block -initial_lower_bound = 16373156 +initial_lower_bound = "" # Janitor sweep interval, should take into account concurrency and queue_size -janitor_sweep_interval = 5 +janitor_sweep_interval = 10 [postgres] -dsn = "postgres://postgres:postgres@localhost:5432/cic_chain_events" +dsn = "" # https://docs.nats.io/ [jetstream] -endpoint = "nats://localhost:4222" +endpoint = "" # Duration JetStream should keep the message before GC -persist_duration_hours = 48 +persist_duration_hrs = 48 # Duration to ignore duplicate transactions (e.g. due to restart) -dedup_duration_hours = 6 +dedup_duration_hrs = 6 diff --git a/docker-compose.yaml b/dev/docker-compose.dev.yaml similarity index 100% rename from docker-compose.yaml rename to dev/docker-compose.dev.yaml diff --git a/docker-compose.build.yaml b/docker-compose.build.yaml deleted file mode 100644 index 4a1e03a..0000000 --- a/docker-compose.build.yaml +++ /dev/null @@ -1,10 +0,0 @@ -version: '3.9' -services: - cic-chain-events: - image: ${IMAGE_BASE_URL:-ghcr.io/grassrootseconomics/cic-chain-events}/cic-chain-events:${TAG:-latest} - build: - context: . - dockerfile: Dockerfile - cache_from: - - ${IMAGE_BASE_URL:-ghcr.io/grassrootseconomics/cic-chain-events}/cic-chain-events:latest - \ No newline at end of file diff --git a/go.mod b/go.mod index 27787f0..150549a 100644 --- a/go.mod +++ b/go.mod @@ -9,8 +9,10 @@ require ( github.com/goccy/go-json v0.10.0 github.com/grassrootseconomics/w3-celo-patch v0.2.0 github.com/jackc/pgx/v5 v5.3.0 + github.com/jackc/tern/v2 v2.0.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/labstack/echo/v4 v4.9.0 github.com/nats-io/nats.go v1.23.0 github.com/stretchr/testify v1.8.1 @@ -19,6 +21,9 @@ require ( require ( filippo.io/edwards25519 v1.0.0-alpha.2 // indirect + github.com/Masterminds/goutils v1.1.1 // indirect + github.com/Masterminds/semver/v3 v3.2.0 // indirect + github.com/Masterminds/sprig/v3 v3.2.3 // indirect github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 // indirect github.com/VictoriaMetrics/fastcache v1.6.0 // indirect github.com/btcsuite/btcd v0.20.1-beta // indirect @@ -36,12 +41,15 @@ require ( github.com/go-ole/go-ole v1.2.5 // indirect github.com/go-stack/stack v1.8.0 // indirect github.com/golang/snappy v0.0.4 // indirect + github.com/google/uuid v1.3.0 // indirect github.com/gorilla/websocket v1.4.2 // indirect github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect github.com/hdevalence/ed25519consensus v0.0.0-20201207055737-7fde80a9d5ff // indirect github.com/holiman/bloomfilter/v2 v2.0.3 // indirect github.com/holiman/uint256 v1.2.1 // indirect + github.com/huandu/xstrings v1.4.0 // indirect github.com/huin/goupnp v1.0.3 // indirect + github.com/imdario/mergo v0.3.13 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect github.com/jackc/puddle/v2 v2.2.0 // indirect @@ -64,6 +72,8 @@ require ( github.com/prometheus/tsdb v0.7.1 // indirect github.com/rogpeppe/go-internal v1.9.0 // indirect github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect + github.com/shopspring/decimal v1.3.1 // indirect + github.com/spf13/cast v1.5.0 // indirect github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect github.com/tklauser/go-sysconf v0.3.5 // indirect github.com/tklauser/numcpus v0.2.2 // indirect diff --git a/go.sum b/go.sum index c87797d..9bc1d83 100644 --- a/go.sum +++ b/go.sum @@ -37,6 +37,12 @@ github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= github.com/DATA-DOG/go-sqlmock v1.3.3/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM= +github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI= +github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU= +github.com/Masterminds/semver/v3 v3.2.0 h1:3MEsd0SM6jqZojhjLWWeBY+Kcjy9i6MQAeY7YgDP83g= +github.com/Masterminds/semver/v3 v3.2.0/go.mod h1:qvl/7zhW3nngYb5+80sSMF+FG2BjYrf8m9wsX0PNOMQ= +github.com/Masterminds/sprig/v3 v3.2.3 h1:eL2fZNezLomi0uOLqjQoN6BfsDD+fyLtgbJMAj9n6YA= +github.com/Masterminds/sprig/v3 v3.2.3/go.mod h1:rXcFaZ2zZbLRJv/xSysmlgIM1u11eBaRMhvYXJNkGuM= github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 h1:fLjPD/aNc3UIOA6tDi6QXUemppXK3P9BI7mr2hd6gx8= github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg= @@ -163,6 +169,7 @@ github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga github.com/fjl/memsize v0.0.0-20190710130421-bcb5799ab5e5 h1:FtmdgXiUlNeRsoNMFlKLDt+S+6hbjVMEW6RGQ7aUf7c= github.com/fjl/memsize v0.0.0-20190710130421-bcb5799ab5e5/go.mod h1:VvhXpOYNQvB+uIk2RvXzuaQtkQJzzIx6lSBe1xv7hi0= github.com/fogleman/gg v1.2.1-0.20190220221249-0403632d5b90/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzPa1k= +github.com/frankban/quicktest v1.14.3 h1:FJKSZTDHjyhriyC81FLQ0LY93eSai0ZyR/ZIkd3ZUKE= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4= github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= @@ -250,9 +257,11 @@ github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OI github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc= github.com/google/pprof v0.0.0-20191218002539-d4f498aebedc/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM= github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI= +github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.1.5/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs= +github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= +github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg= github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk= github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= @@ -292,7 +301,6 @@ github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= 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/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64= github.com/hashicorp/mdns v1.0.4/go.mod h1:mtBihi+LeNXGtG8L9dX59gAEa12BDtBQSp4v/YAJqrc= @@ -304,7 +312,6 @@ github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKe github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM= github.com/hdevalence/ed25519consensus v0.0.0-20201207055737-7fde80a9d5ff h1:LeVKjw8pcDQj7WVVnbFvbD7ovcv+r/l15ka1NH6Lswc= github.com/hdevalence/ed25519consensus v0.0.0-20201207055737-7fde80a9d5ff/go.mod h1:Feit0l8NcNO4g69XNjwvsR0LGcwMMfzI1TF253rOIlQ= -github.com/hjson/hjson-go/v4 v4.0.0 h1:wlm6IYYqHjOdXH1gHev4VoXCaW20HdQAGCxdOEEg2cs= github.com/hjson/hjson-go/v4 v4.0.0/go.mod h1:KaYt3bTw3zhBjYqnXkYywcYctk0A2nxeEFTse3rH13E= github.com/holiman/bloomfilter/v2 v2.0.3 h1:73e0e/V0tCydx14a0SCYS/EWCxgwLZ18CZcZKVu0fao= github.com/holiman/bloomfilter/v2 v2.0.3/go.mod h1:zpoh+gs7qcpqrHr3dB55AMiJwo0iURXE7ZOP9L9hSkA= @@ -312,11 +319,17 @@ github.com/holiman/uint256 v1.2.0/go.mod h1:y4ga/t+u+Xwd7CpDgZESaRcWy0I7XMlTMA25 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/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= +github.com/huandu/xstrings v1.3.3/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE= +github.com/huandu/xstrings v1.4.0 h1:D17IlohoQq4UcpqD7fDk80P7l+lwAmlFaBHgOipl2FU= +github.com/huandu/xstrings v1.4.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE= github.com/huin/goupnp v1.0.2/go.mod h1:0dxJBVBHqTMjIUMkESDTNgOOx/Mw5wYIfyFmdzSamkM= github.com/huin/goupnp v1.0.3 h1:N8No57ls+MnjlB+JPiCVSOyy/ot7MJTqlo7rn+NYSqQ= github.com/huin/goupnp v1.0.3/go.mod h1:ZxNlw5WqJj6wSsRK5+YfflQGXYfccj5VgQsMNixHM7Y= github.com/huin/goutil v0.0.0-20170803182201-1ca381bf3150/go.mod h1:PpLOETDnJ0o3iZrZfqZzyLl6l7F3c6L1oWn7OICBi6o= github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= +github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA= +github.com/imdario/mergo v0.3.13 h1:lFzP57bqS/wsqKssCGmtLAb8A0wKjLGrve2q3PPVcBk= +github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg= github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= github.com/influxdata/flux v0.65.1/go.mod h1:J754/zds0vvpfwuq7Gc2wRdVwEodfpCFM7mYlOw2LqY= github.com/influxdata/influxdb v1.8.3/go.mod h1:JugdFhsvvI8gadxOI6noqNeeBHvWNTbfYGtiAn+2jhI= @@ -337,6 +350,8 @@ github.com/jackc/pgx/v5 v5.3.0 h1:/NQi8KHMpKWHInxXesC8yD4DhkXPrVhmnwYkjp9AmBA= github.com/jackc/pgx/v5 v5.3.0/go.mod h1:t3JDKnCBlYIc0ewLF0Q7B8MXmoIaBOZj/ic7iHozM/8= 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.0 h1:ZE3nUQdjGFljIB2ExOgh9/2snUBfzvbAlbP8jt92xI0= +github.com/jackc/tern/v2 v2.0.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= @@ -345,7 +360,6 @@ github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M 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/jmoiron/sqlx v1.2.0/go.mod h1:1FEQNm3xlJgrMD+FBdI9+xvCksHtbpVBBw5dYhBSsks= -github.com/joho/godotenv v1.3.0 h1:Zjp+RcGpHhGlrMbJzXTrZZPrWj+1vfm90La1wgB6Bhc= github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg= github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ= @@ -374,6 +388,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.0/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= @@ -381,8 +397,8 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= -github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= 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/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/labstack/echo/v4 v4.2.1/go.mod h1:AA49e0DZ8kk5jTOOCKNuPR6oTnBS0dYiM4FW1e6jwpg= github.com/labstack/echo/v4 v4.9.0 h1:wPOF1CE6gvt/kmbMR4dGzWvHMPT+sAEUJOwOTtvITVY= @@ -542,6 +558,8 @@ github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAm github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible h1:Bn1aCHHRnjv4Bl16T8rcaFjYSrGrIZvpiGO6P3Q4GpU= github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o= +github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8= +github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= @@ -550,9 +568,11 @@ github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE= +github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE= +github.com/spf13/cast v1.5.0 h1:rj3WzYc11XZaIZMPKmwP96zkFEnnAmV8s6XbB2aY32w= +github.com/spf13/cast v1.5.0/go.mod h1:SpXXQ5YoyJw6s3/6cMTQuxvgRl3PCJiyaX9p6b155UU= github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ= github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= -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.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -591,6 +611,7 @@ github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6/go.mod h1:ce1O1j6Ut github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/zerodha/logf v0.5.5 h1:AhxHlixHNYwhFjvlgTv6uO4VBKYKxx2I6SbHoHtWLBk= github.com/zerodha/logf v0.5.5/go.mod h1:HWpfKsie+WFFpnUnUxelT6Z0FC6xu9+qt+oXNMPg6y8= go.etcd.io/etcd/api/v3 v3.5.4/go.mod h1:5GB2vv4A4AOn3yk7MftYGHkUfGtDHnEraIjym4dYz5A= @@ -619,6 +640,8 @@ golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPh 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.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= @@ -651,6 +674,7 @@ golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzB golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -679,6 +703,8 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod h1:9tjilg8BloeKEkVJvy7fQ90B1CfIiPueXVOjqfkSzI8= 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.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= @@ -696,6 +722,7 @@ golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -751,15 +778,21 @@ golang.org/x/sys v0.0.0-20210420205809-ac73e9fd8988/go.mod h1:h1NjWce9XRLGQEsW7w 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= golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211103235746-7861aae1554b/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.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.5.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/term v0.2.0/go.mod h1:TVmDHMZPmdnySmBfhjOoOdhjzdE1h4u1VwSiw2l1Nuc= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20181227161524-e6919f6577db/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= @@ -768,6 +801,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.4/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/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= @@ -808,6 +843,7 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0= golang.org/x/tools v0.1.2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -900,6 +936,7 @@ gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= 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.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo= diff --git a/internal/events/events.go b/internal/events/events.go deleted file mode 100644 index 02c2bc7..0000000 --- a/internal/events/events.go +++ /dev/null @@ -1,17 +0,0 @@ -package events - -type EventEmitter interface { - Close() - Publish(subject string, dedupId string, eventPayload interface{}) error -} - -type MinimalTxInfo struct { - Block uint64 `json:"block"` - From string `json:"from"` - To string `json:"to"` - ContractAddress string `json:"contractAddress"` - Success bool `json:"success"` - TxHash string `json:"transactionHash"` - TxIndex uint `json:"transactionIndex"` - Value uint64 `json:"value"` -} diff --git a/internal/events/jetstream.go b/internal/events/jetstream.go deleted file mode 100644 index 31bbd40..0000000 --- a/internal/events/jetstream.go +++ /dev/null @@ -1,78 +0,0 @@ -package events - -import ( - "time" - - "github.com/goccy/go-json" - "github.com/nats-io/nats.go" -) - -const ( - StreamName string = "CHAIN" - StreamSubjects string = "CHAIN.*" -) - -type JetStreamOpts struct { - ServerUrl string - PersistDuration time.Duration - DedupDuration time.Duration -} - -type JetStream struct { - jsCtx nats.JetStreamContext - nc *nats.Conn -} - -func NewJetStreamEventEmitter(o JetStreamOpts) (EventEmitter, error) { - natsConn, err := nats.Connect(o.ServerUrl) - if err != nil { - return nil, err - } - - js, err := natsConn.JetStream() - if err != nil { - return nil, err - } - - // Bootstrap stream if it doesn't exist. - stream, _ := js.StreamInfo(StreamName) - if stream == nil { - _, err = js.AddStream(&nats.StreamConfig{ - Name: StreamName, - MaxAge: o.PersistDuration, - Storage: nats.FileStorage, - Subjects: []string{StreamSubjects}, - Duplicates: o.DedupDuration, - }) - if err != nil { - return nil, err - } - } - - return &JetStream{ - jsCtx: js, - nc: natsConn, - }, nil -} - -// Close gracefully shutdowns the JetStream connection. -func (js *JetStream) Close() { - if js.nc != nil { - js.nc.Close() - } -} - -// Publish publishes the JSON data to the NATS stream. -func (js *JetStream) Publish(subject string, dedupId string, eventPayload interface{}) error { - jsonData, err := json.Marshal(eventPayload) - if err != nil { - return err - } - - _, err = js.jsCtx.Publish(subject, jsonData, nats.MsgId(dedupId)) - if err != nil { - return err - } - - return nil -} diff --git a/internal/filter/address_filter.go b/internal/filter/address_filter.go index 00f4d74..74c3f7f 100644 --- a/internal/filter/address_filter.go +++ b/internal/filter/address_filter.go @@ -8,17 +8,19 @@ import ( "github.com/zerodha/logf" ) -type AddressFilterOpts struct { - Cache *sync.Map - Logg logf.Logger - SystemAddress string -} +type ( + AddressFilterOpts struct { + Cache *sync.Map + Logg logf.Logger + SystemAddress string + } -type AddressFilter struct { - cache *sync.Map - logg logf.Logger - systemAddress string -} + AddressFilter struct { + cache *sync.Map + logg logf.Logger + systemAddress string + } +) func NewAddressFilter(o AddressFilterOpts) Filter { return &AddressFilter{ diff --git a/internal/filter/gas_filter.go b/internal/filter/gas_filter.go index e164d52..add861c 100644 --- a/internal/filter/gas_filter.go +++ b/internal/filter/gas_filter.go @@ -4,7 +4,7 @@ import ( "context" "github.com/celo-org/celo-blockchain/common/hexutil" - "github.com/grassrootseconomics/cic-chain-events/internal/events" + "github.com/grassrootseconomics/cic-chain-events/internal/pub" "github.com/grassrootseconomics/cic-chain-events/pkg/fetch" "github.com/zerodha/logf" ) @@ -13,22 +13,24 @@ const ( gasFilterEventSubject = "CHAIN.gas" ) -type GasFilterOpts struct { - EventEmitter events.EventEmitter - Logg logf.Logger - SystemAddress string -} +type ( + GasFilterOpts struct { + Logg logf.Logger + Pub *pub.Pub + SystemAddress string + } -type GasFilter struct { - eventEmitter events.EventEmitter - logg logf.Logger - systemAddress string -} + GasFilter struct { + logg logf.Logger + pub *pub.Pub + systemAddress string + } +) func NewGasFilter(o GasFilterOpts) Filter { return &GasFilter{ - eventEmitter: o.EventEmitter, logg: o.Logg, + pub: o.Pub, systemAddress: o.SystemAddress, } } @@ -41,7 +43,7 @@ func (f *GasFilter) Execute(_ context.Context, transaction fetch.Transaction) (b // TODO: This is a temporary shortcut to gift gas. Switch to gas faucet contract. if transaction.From.Address == f.systemAddress && transferValue > 0 { - transferEvent := &events.MinimalTxInfo{ + transferEvent := &pub.MinimalTxInfo{ Block: transaction.Block.Number, To: transaction.To.Address, TxHash: transaction.Hash, @@ -53,7 +55,7 @@ func (f *GasFilter) Execute(_ context.Context, transaction fetch.Transaction) (b transferEvent.Success = true } - if err := f.eventEmitter.Publish( + if err := f.pub.Publish( gasFilterEventSubject, transaction.Hash, transferEvent, diff --git a/internal/filter/register_filter.go b/internal/filter/register_filter.go index 093dcea..a6ddcfe 100644 --- a/internal/filter/register_filter.go +++ b/internal/filter/register_filter.go @@ -4,7 +4,7 @@ import ( "context" "github.com/celo-org/celo-blockchain/common" - "github.com/grassrootseconomics/cic-chain-events/internal/events" + "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" @@ -18,20 +18,22 @@ var ( addSig = w3.MustNewFunc("add(address)", "bool") ) -type RegisterFilterOpts struct { - EventEmitter events.EventEmitter - Logg logf.Logger -} +type ( + RegisterFilterOpts struct { + Logg logf.Logger + Pub *pub.Pub + } -type RegisterFilter struct { - eventEmitter events.EventEmitter - logg logf.Logger -} + RegisterFilter struct { + logg logf.Logger + pub *pub.Pub + } +) func NewRegisterFilter(o RegisterFilterOpts) Filter { return &RegisterFilter{ - eventEmitter: o.EventEmitter, - logg: o.Logg, + logg: o.Logg, + pub: o.Pub, } } @@ -47,7 +49,7 @@ func (f *RegisterFilter) Execute(_ context.Context, transaction fetch.Transactio return false, err } - addEvent := &events.MinimalTxInfo{ + addEvent := &pub.MinimalTxInfo{ Block: transaction.Block.Number, ContractAddress: transaction.To.Address, To: transaction.To.Address, @@ -59,7 +61,7 @@ func (f *RegisterFilter) Execute(_ context.Context, transaction fetch.Transactio addEvent.Success = true } - if err := f.eventEmitter.Publish( + if err := f.pub.Publish( registerEventSubject, transaction.Hash, addEvent, @@ -69,6 +71,6 @@ func (f *RegisterFilter) Execute(_ context.Context, transaction fetch.Transactio return true, nil } - + return true, nil } diff --git a/internal/filter/transfer_filter.go b/internal/filter/transfer_filter.go index 5bbe30a..26c2eeb 100644 --- a/internal/filter/transfer_filter.go +++ b/internal/filter/transfer_filter.go @@ -5,7 +5,7 @@ import ( "math/big" "github.com/celo-org/celo-blockchain/common" - "github.com/grassrootseconomics/cic-chain-events/internal/events" + "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" @@ -21,20 +21,22 @@ var ( mintToSig = w3.MustNewFunc("mintTo(address, uint256)", "bool") ) -type TransferFilterOpts struct { - EventEmitter events.EventEmitter - Logg logf.Logger -} +type ( + TransferFilterOpts struct { + Logg logf.Logger + Pub *pub.Pub + } -type TransferFilter struct { - eventEmitter events.EventEmitter - logg logf.Logger -} + TransferFilter struct { + logg logf.Logger + pub *pub.Pub + } +) func NewTransferFilter(o TransferFilterOpts) Filter { return &TransferFilter{ - eventEmitter: o.EventEmitter, - logg: o.Logg, + logg: o.Logg, + pub: o.Pub, } } @@ -56,7 +58,7 @@ func (f *TransferFilter) Execute(_ context.Context, transaction fetch.Transactio f.logg.Debug("transfer_filter: new reg", "transfer", to) - transferEvent := &events.MinimalTxInfo{ + transferEvent := &pub.MinimalTxInfo{ Block: transaction.Block.Number, From: transaction.From.Address, To: to.Hex(), @@ -70,7 +72,7 @@ func (f *TransferFilter) Execute(_ context.Context, transaction fetch.Transactio transferEvent.Success = true } - if err := f.eventEmitter.Publish( + if err := f.pub.Publish( transferFilterEventSubject, transaction.Hash, transferEvent, @@ -92,7 +94,7 @@ func (f *TransferFilter) Execute(_ context.Context, transaction fetch.Transactio f.logg.Debug("transfer_filter: new reg", "transferFrom", to) - transferEvent := &events.MinimalTxInfo{ + transferEvent := &pub.MinimalTxInfo{ Block: transaction.Block.Number, From: from.Hex(), To: to.Hex(), @@ -106,7 +108,7 @@ func (f *TransferFilter) Execute(_ context.Context, transaction fetch.Transactio transferEvent.Success = true } - if err := f.eventEmitter.Publish( + if err := f.pub.Publish( transferFilterEventSubject, transaction.Hash, transferEvent, @@ -127,7 +129,7 @@ func (f *TransferFilter) Execute(_ context.Context, transaction fetch.Transactio f.logg.Debug("transfer_filter: new reg", "mintTo", to) - transferEvent := &events.MinimalTxInfo{ + transferEvent := &pub.MinimalTxInfo{ Block: transaction.Block.Number, From: transaction.From.Address, To: to.Hex(), @@ -141,7 +143,7 @@ func (f *TransferFilter) Execute(_ context.Context, transaction fetch.Transactio transferEvent.Success = true } - if err := f.eventEmitter.Publish( + if err := f.pub.Publish( transferFilterEventSubject, transaction.Hash, transferEvent, diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 78872ff..c4ffec4 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -10,19 +10,21 @@ import ( "github.com/zerodha/logf" ) -type PipelineOpts struct { - BlockFetcher fetch.Fetch - Filters []filter.Filter - Logg logf.Logger - Store store.Store[pgx.Rows] -} +type ( + PipelineOpts struct { + BlockFetcher fetch.Fetch + Filters []filter.Filter + Logg logf.Logger + Store store.Store[pgx.Rows] + } -type Pipeline struct { - fetch fetch.Fetch - filters []filter.Filter - logg logf.Logger - store store.Store[pgx.Rows] -} + Pipeline struct { + fetch fetch.Fetch + filters []filter.Filter + logg logf.Logger + store store.Store[pgx.Rows] + } +) func NewPipeline(o PipelineOpts) *Pipeline { return &Pipeline{ diff --git a/internal/pool/pool.go b/internal/pool/pool.go index cc5f9f5..6e46cfd 100644 --- a/internal/pool/pool.go +++ b/internal/pool/pool.go @@ -7,6 +7,10 @@ import ( "github.com/alitto/pond" ) +const ( + idleTimeout = 1 * time.Second +) + type Opts struct { Concurrency int QueueSize int @@ -18,7 +22,7 @@ func NewPool(ctx context.Context, o Opts) *pond.WorkerPool { o.Concurrency, o.QueueSize, pond.MinWorkers(o.Concurrency), - pond.IdleTimeout(time.Second*1), + pond.IdleTimeout(idleTimeout), pond.Context(ctx), ) } diff --git a/internal/pub/jetstream.go b/internal/pub/jetstream.go new file mode 100644 index 0000000..43c58e3 --- /dev/null +++ b/internal/pub/jetstream.go @@ -0,0 +1,81 @@ +package pub + +import ( + "time" + + "github.com/goccy/go-json" + "github.com/nats-io/nats.go" +) + +const ( + streamName string = "CHAIN" + streamSubjects string = "CHAIN.*" +) + +type ( + PubOpts struct { + DedupDuration time.Duration + JsCtx nats.JetStreamContext + NatsConn *nats.Conn + PersistDuration time.Duration + } + + Pub struct { + natsConn *nats.Conn + jsCtx nats.JetStreamContext + } + + MinimalTxInfo struct { + Block uint64 `json:"block"` + From string `json:"from"` + To string `json:"to"` + ContractAddress string `json:"contractAddress"` + Success bool `json:"success"` + TxHash string `json:"transactionHash"` + TxIndex uint `json:"transactionIndex"` + Value uint64 `json:"value"` + } +) + +func NewPub(o PubOpts) (*Pub, error) { + stream, _ := o.JsCtx.StreamInfo(streamName) + if stream == nil { + _, err := o.JsCtx.AddStream(&nats.StreamConfig{ + Name: streamName, + MaxAge: o.PersistDuration, + Storage: nats.FileStorage, + Subjects: []string{streamSubjects}, + Duplicates: o.DedupDuration, + }) + if err != nil { + return nil, err + } + } + + return &Pub{ + jsCtx: o.JsCtx, + natsConn: o.NatsConn, + }, nil +} + +// Close gracefully shutdowns the JetStream connection. +func (p *Pub) Close() { + if p.natsConn != nil { + p.natsConn.Close() + } +} + +// Publish publishes the JSON data to the NATS stream. +func (p *Pub) Publish(subject string, dedupId string, eventPayload interface{}) error { + jsonData, err := json.Marshal(eventPayload) + if err != nil { + return err + } + + _, err = p.jsCtx.Publish(subject, jsonData, nats.MsgId(dedupId)) + if err != nil { + return err + } + + return nil +} diff --git a/internal/store/postgres.go b/internal/store/postgres.go index 233cb0b..e1f826e 100644 --- a/internal/store/postgres.go +++ b/internal/store/postgres.go @@ -3,27 +3,37 @@ package store import ( "context" "fmt" + "os" + "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + "github.com/jackc/tern/v2/migrate" "github.com/knadh/goyesql/v2" "github.com/zerodha/logf" ) -type queries struct { - CommitBlock string `query:"commit-block"` - GetMissingBlocks string `query:"get-missing-blocks"` - GetSearchBounds string `query:"get-search-bounds"` - InitSyncerMeta string `query:"init-syncer-meta"` - SetSearchLowerBound string `query:"set-search-lower-bound"` -} +const ( + schemaTable = "schema_version" +) -type PostgresStoreOpts struct { - DSN string - InitialLowerBound uint64 - Logg logf.Logger - Queries goyesql.Queries -} +type ( + queries struct { + CommitBlock string `query:"commit-block"` + GetMissingBlocks string `query:"get-missing-blocks"` + GetSearchBounds string `query:"get-search-bounds"` + InitSyncerMeta string `query:"init-syncer-meta"` + SetSearchLowerBound string `query:"set-search-lower-bound"` + } + + PostgresStoreOpts struct { + DSN string + MigrationsFolderPath string + InitialLowerBound uint64 + Logg logf.Logger + Queries goyesql.Queries + } +) type PostgresStore struct { logg logf.Logger @@ -45,12 +55,34 @@ func NewPostgresStore(o PostgresStoreOpts) (Store[pgx.Rows], error) { return nil, err } - dbPool, err := pgxpool.NewWithConfig(context.Background(), parsedConfig) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + dbPool, err := pgxpool.NewWithConfig(ctx, parsedConfig) if err != nil { return nil, err } - _, err = dbPool.Exec(context.Background(), postgresStore.queries.InitSyncerMeta, o.InitialLowerBound) + conn, err := dbPool.Acquire(ctx) + if err != nil { + return nil, err + } + defer conn.Release() + + migrator, err := migrate.NewMigrator(ctx, conn.Conn(), schemaTable) + if err != nil { + return nil, err + } + + if err := migrator.LoadMigrations(os.DirFS(o.MigrationsFolderPath)); err != nil { + return nil, err + } + + if err := migrator.Migrate(ctx); err != nil { + return nil, err + } + + _, err = dbPool.Exec(ctx, postgresStore.queries.InitSyncerMeta, o.InitialLowerBound) if err != nil { return nil, err } diff --git a/internal/syncer/head.go b/internal/syncer/head.go index dad70ae..818ca24 100644 --- a/internal/syncer/head.go +++ b/internal/syncer/head.go @@ -2,6 +2,7 @@ package syncer import ( "context" + "time" "github.com/alitto/pond" "github.com/celo-org/celo-blockchain/core/types" @@ -10,21 +11,27 @@ import ( "github.com/zerodha/logf" ) -type HeadSyncerOpts struct { - Stats *Stats - Pipeline *pipeline.Pipeline - Logg logf.Logger - Pool *pond.WorkerPool - WsEndpoint string -} +const ( + jobTimeout = 5 * time.Second +) -type HeadSyncer struct { - stats *Stats - pipeline *pipeline.Pipeline - logg logf.Logger - ethClient *ethclient.Client - pool *pond.WorkerPool -} +type ( + HeadSyncerOpts struct { + Logg logf.Logger + Pipeline *pipeline.Pipeline + Pool *pond.WorkerPool + Stats *Stats + WsEndpoint string + } + + HeadSyncer struct { + ethClient *ethclient.Client + logg logf.Logger + pipeline *pipeline.Pipeline + pool *pond.WorkerPool + stats *Stats + } +) func NewHeadSyncer(o HeadSyncerOpts) (*HeadSyncer, error) { ethClient, err := ethclient.Dial(o.WsEndpoint) @@ -33,11 +40,11 @@ func NewHeadSyncer(o HeadSyncerOpts) (*HeadSyncer, error) { } return &HeadSyncer{ - stats: o.Stats, - pipeline: o.Pipeline, - logg: o.Logg, ethClient: ethClient, + logg: o.Logg, + pipeline: o.Pipeline, pool: o.Pool, + stats: o.Stats, }, nil } @@ -46,7 +53,7 @@ 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(ctx, headerReceiver) + sub, err := hs.ethClient.SubscribeNewHead(context.Background(), headerReceiver) if err != nil { return err } @@ -62,10 +69,12 @@ func (hs *HeadSyncer) Start(ctx context.Context) error { case header := <-headerReceiver: blockNumber := header.Number.Uint64() hs.logg.Debug("head syncer: received new block", "block", blockNumber) - hs.stats.UpdateHeadCursor(blockNumber) hs.pool.Submit(func() { - if err := hs.pipeline.Run(context.Background(), blockNumber); err != nil { + ctx, cancel := context.WithTimeout(context.Background(), jobTimeout) + defer cancel() + + if err := hs.pipeline.Run(ctx, blockNumber); err != nil { hs.logg.Error("head syncer: piepline run error", "error", err) } }) diff --git a/internal/syncer/janitor.go b/internal/syncer/janitor.go index 1a3cd63..d071fc7 100644 --- a/internal/syncer/janitor.go +++ b/internal/syncer/janitor.go @@ -15,25 +15,27 @@ const ( headBlockLag = 5 ) -type JanitorOpts struct { - BatchSize uint64 - Logg logf.Logger - Pipeline *pipeline.Pipeline - Pool *pond.WorkerPool - Stats *Stats - Store store.Store[pgx.Rows] - SweepInterval time.Duration -} +type ( + JanitorOpts struct { + BatchSize uint64 + Logg logf.Logger + Pipeline *pipeline.Pipeline + Pool *pond.WorkerPool + Stats *Stats + Store store.Store[pgx.Rows] + SweepInterval time.Duration + } -type Janitor struct { - batchSize uint64 - pipeline *pipeline.Pipeline - logg logf.Logger - pool *pond.WorkerPool - stats *Stats - store store.Store[pgx.Rows] - sweepInterval time.Duration -} + Janitor struct { + batchSize uint64 + pipeline *pipeline.Pipeline + logg logf.Logger + pool *pond.WorkerPool + stats *Stats + store store.Store[pgx.Rows] + sweepInterval time.Duration + } +) func NewJanitor(o JanitorOpts) *Janitor { return &Janitor{ @@ -56,8 +58,11 @@ func (j *Janitor) Start(ctx context.Context) error { j.logg.Info("janitor: shutdown signal received") return nil case <-ticker.C: + ctx, cancel := context.WithTimeout(context.Background(), jobTimeout) + defer cancel() + j.logg.Debug("janitor: starting sweep") - if err := j.QueueMissingBlocks(context.Background()); err != nil { + if err := j.QueueMissingBlocks(ctx); err != nil { j.logg.Error("janitor: queue missing blocks error", "error", err) } }