Rewrite fetcher logic to be more robust

.gitignore

@@ -1,6 +1,9 @@
 # MacOS Leftovers
 .DS_Store
 
+# Data Leftovers
+indexer-db/*
+
 # Editor Leftovers
 .vscode
 .idea

client/http.go

@@ -0,0 +1,47 @@
+package client
+
+import (
+ "fmt"
+
+ rpcClient "github.com/gnolang/gno/tm2/pkg/bft/rpc/client"
+ core_types "github.com/gnolang/gno/tm2/pkg/bft/rpc/core/types"
+)
+
+// Client is the TM2 HTTP client
+type Client struct {
+ client rpcClient.Client
+}
+
+// NewClient creates a new TM2 HTTP client
+func NewClient(remote string) *Client {
+ return &Client{
+ client: rpcClient.NewHTTP(remote, ""),
+ }
+}
+
+func (c *Client) GetLatestBlockNumber() (int64, error) {
+ status, err := c.client.Status()
+ if err != nil {
+ return 0, fmt.Errorf("unable to get chain status, %w", err)
+ }
+
+ return status.SyncInfo.LatestBlockHeight, nil
+}
+
+func (c *Client) GetBlock(blockNum int64) (*core_types.ResultBlock, error) {
+ block, err := c.client.Block(&blockNum)
+ if err != nil {
+ return nil, fmt.Errorf("unable to get block, %w", err)
+ }
+
+ return block, nil
+}
+
+func (c *Client) GetBlockResults(blockNum int64) (*core_types.ResultBlockResults, error) {
+ results, err := c.client.BlockResults(&blockNum)
+ if err != nil {
+ return nil, fmt.Errorf("unable to get block results, %w", err)
+ }
+
+ return results, nil
+}

cmd/start.go

@@ -3,11 +3,26 @@ package main
 import (
  "context"
  "flag"
+ "fmt"
 
+ "github.com/gnolang/tx-indexer/client"
+ "github.com/gnolang/tx-indexer/fetch"
+ "github.com/gnolang/tx-indexer/serve"
+ "github.com/gnolang/tx-indexer/storage"
  "github.com/peterbourgon/ff/v3/ffcli"
+ "go.uber.org/zap"
 )
 
-type startCfg struct{}
+const (
+ defaultRemote = "http://127.0.0.1:26657"
+ defaultDBPath = "indexer-db"
+)
+
+type startCfg struct {
+ listenAddress string
+ remote string
+ dbPath string
+}
 
 // newStartCmd creates the indexer start command
 func newStartCmd() *ffcli.Command {
@@ -19,19 +34,87 @@ func newStartCmd() *ffcli.Command {
  return &ffcli.Command{
  Name: "start",
  ShortUsage: "start [flags]",
- LongHelp: "Starts the indexer",
+ ShortHelp: "Starts the indexer service",
+ LongHelp: "Starts the indexer service, which includes the fetcher and JSON-RPC server",
  FlagSet: fs,
- Exec: cfg.exec,
+ Exec: func(ctx context.Context, _ []string) error {
+ return cfg.exec(ctx)
+ },
  }
 }
 
 // registerFlags registers the indexer start command flags
-func (c *startCfg) registerFlags(_ *flag.FlagSet) {
- // TODO define flags
+func (c *startCfg) registerFlags(fs *flag.FlagSet) {
+ fs.StringVar(
+ &c.listenAddress,
+ "listen-address",
+ serve.DefaultListenAddress,
+ "the IP:PORT URL for the indexer JSON-RPC server",
+ )
+
+ fs.StringVar(
+ &c.remote,
+ "remote",
+ defaultRemote,
+ "the JSON-RPC URL of the Gno chain",
+ )
+
+ fs.StringVar(
+ &c.dbPath,
+ "db-path",
+ defaultDBPath,
+ "the absolute path for the indexer DB (embedded)",
+ )
 }
 
 // exec executes the indexer start command
-func (c *startCfg) exec(_ context.Context, _ []string) error {
- // TODO add implementation
- return nil
+func (c *startCfg) exec(ctx context.Context) error {
+ // Create a new logger
+ logger, err := zap.NewDevelopment()
+ if err != nil {
+ return fmt.Errorf("unable to create logger, %w", err)
+ }
+
+ // Create a DB instance
+ db, err := storage.New(c.dbPath)
+ if err != nil {
+ return fmt.Errorf("unable to open storage DB, %w", err)
+ }
+
+ defer func() {
+ if err := db.Close(); err != nil {
+ logger.Error("unable to gracefully close DB", zap.Error(err))
+ }
+ }()
+
+ // Create the fetcher instance
+ f := fetch.New(
+ db,
+ client.NewClient(c.remote),
+ fetch.WithLogger(
+ logger.Named("fetcher"),
+ ),
+ )
+
+ // Create the JSON-RPC service
+ j := serve.NewJSONRPC(
+ serve.WithLogger(
+ logger.Named("json-rpc"),
+ ),
+ serve.WithListenAddress(
+ c.listenAddress,
+ ),
+ )
+
+ // Create a new waiter
+ w := newWaiter(ctx)
+
+ // Add the fetcher service
+ w.add(f.FetchTransactions)
+
+ // Add the JSON-RPC service
+ w.add(j.Serve)
+
+ // Wait for the services to stop
+ return w.wait()
 }

cmd/waiter.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+ "context"
+ "os"
+ "os/signal"
+ "syscall"
+
+ "golang.org/x/sync/errgroup"
+)
+
+type waitFunc func(ctx context.Context) error
+
+// waiter is a concept used for waiting on running services
+type waiter struct {
+ ctx context.Context
+ cancel context.CancelFunc
+
+ waitFns []waitFunc
+}
+
+// newWaiter creates a new waiter instance
+func newWaiter(ctx context.Context) *waiter {
+ w := &waiter{
+ waitFns: []waitFunc{},
+ }
+
+ w.ctx, w.cancel = signal.NotifyContext(
+ ctx,
+ os.Interrupt,
+ syscall.SIGINT,
+ syscall.SIGTERM,
+ syscall.SIGQUIT,
+ )
+
+ return w
+}
+
+// add adds a new wait service
+func (w *waiter) add(fns ...waitFunc) {
+ w.waitFns = append(w.waitFns, fns...)
+}
+
+// wait blocks until all added wait services finish
+func (w *waiter) wait() error {
+ g, ctx := errgroup.WithContext(w.ctx)
+
+ g.Go(func() error {
+ <-ctx.Done()
+ w.cancel()
+
+ return nil
+ })
+
+ for _, fn := range w.waitFns {
+ fn := fn
+
+ g.Go(
+ func() error {
+ return fn(ctx)
+ },
+ )
+ }
+
+ return g.Wait()
+}