tracker

README

Tracker

package main

import (
	"context"
	"encoding/binary"
	"flag"
	"fmt"
	"os"
	"os/signal"
	"syscall"

	web3 "github.com/jx-im/erc-sdk"
	"github.com/jx-im/erc-sdk/abi"
	"github.com/jx-im/erc-sdk/jsonrpc"
	"github.com/jx-im/erc-sdk/tracker"

	boltdbStore "github.com/jx-im/erc-sdk/tracker/store/boltdb"
)

func main() {
	var endpoint string
	var target string

	flag.StringVar(&endpoint, "endpoint", "", "")
	flag.StringVar(&target, "target", "", "")

	flag.Parse()

	provider, err := jsonrpc.NewClient(endpoint)
	if err != nil {
		fmt.Printf("[ERR]: %v", err)
		os.Exit(1)
	}

	tConfig := tracker.DefaultConfig()
	tConfig.BatchSize = 2000
	tConfig.EtherscanFastTrack = true

	store, err := boltdbStore.New("db.db")
	if err != nil {
		fmt.Printf("[ERR]: %v", err)
		os.Exit(1)
	}

	t := tracker.NewTracker(provider.Eth(), tConfig)
	t.SetStore(store)

	ctx, cancelFn := context.WithCancel(context.Background())
	go func() {
		if err := start(ctx, t, web3.HexToAddress(target)); err != nil {
			fmt.Printf("[ERR]: %v", err)
		}
	}()

	handleSignals(cancelFn)
}

var depositContract = `
tuple (
	bytes pubkey,
	bytes whitdrawalcred,
	bytes amount,
	bytes signature,
	bytes index
)
`

func start(ctx context.Context, t *tracker.Tracker, targetAddr web3.Address) error {
	if err := t.Start(ctx); err != nil {
		return err
	}

	fmt.Println("Tracker is ready")

	depositEvent := abi.NewEvent("DepositEvent", abi.MustNewType(depositContract))

	// create the filter
	fConfig := &tracker.FilterConfig{
		Async: true,
		Address: []web3.Address{
			targetAddr,
		},
	}
	f, err := t.NewFilter(fConfig)
	if err != nil {
		return err
	}

	go func() {
		for {
			evnt := <-f.EventCh
			for _, log := range evnt.Added {
				if depositEvent.Match(log) {
					vals, err := depositEvent.ParseLog(log)
					if err != nil {
						panic(err)
					}

					index := binary.LittleEndian.Uint64(vals["index"].([]byte))
					amount := binary.LittleEndian.Uint64(vals["amount"].([]byte))

					fmt.Printf("Deposit: Block %d Index %d Amount %d\n", log.BlockNumber, index, amount)
				}
			}
		}
	}()

	if err := f.Sync(ctx); err != nil {
		return err
	}

	fmt.Println("Historical sync is done")
	return nil
}

func handleSignals(cancelFn context.CancelFunc) int {
	signalCh := make(chan os.Signal, 4)
	signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP)

	<-signalCh

	gracefulCh := make(chan struct{})
	go func() {
		cancelFn()
		close(gracefulCh)
	}()

	select {
	case <-signalCh:
		return 1
	case <-gracefulCh:
		return 0
	}
}

You can query the Prysmatic labs Testnet ETH2.0 Deposit contract like so:

go run main.go --endpoint https://goerli.infura.io/v3/... --target 0x4689a3C63CE249355C8a573B5974db21D2d1b8Ef

Documentation

Index

• type BlockEvent
• type BlockTracker
• type Config
  • func DefaultConfig() *Config
• type Event
• type EventType
• type Filter
  • func (f *Filter) Entry() store.Entry
  • func (f *Filter) GetLastBlock() (*web3.Block, error)
  • func (f *Filter) IsSynced() bool
  • func (f *Filter) Sync(ctx context.Context) error
  • func (f *Filter) SyncAsync(ctx context.Context)
  • func (f *Filter) Wait()
  • func (f *Filter) WaitDuration(dur time.Duration) error
• type FilterConfig
• type JSONBlockTracker
  • func NewJSONBlockTracker(logger *log.Logger, provider Provider) *JSONBlockTracker
  • func (k *JSONBlockTracker) Track(ctx context.Context, handle func(block *web3.Block) error) error
• type Provider
• type SubscriptionBlockTracker
  • func NewSubscriptionBlockTracker(logger *log.Logger, client *jsonrpc.Client) (*SubscriptionBlockTracker, error)
  • func (s *SubscriptionBlockTracker) Track(ctx context.Context, handle func(block *web3.Block) error) error
• type Tracker
  • func NewTracker(provider Provider, config *Config) *Tracker
  • func (t *Tracker) GetSavedFilters() ([]*FilterConfig, error)
  • func (t *Tracker) NewFilter(config *FilterConfig) (*Filter, error)
  • func (t *Tracker) SetLogger(logger *log.Logger)
  • func (t *Tracker) SetStore(store store.Store)
  • func (t *Tracker) Start(ctx context.Context) error
  • func (t *Tracker) Sync(ctx context.Context, filter *Filter) error
  • func (t *Tracker) SyncAsync(ctx context.Context, filter *Filter)

Types

type BlockEvent

type BlockEvent struct {
	Type    EventType
	Added   []*web3.Block
	Removed []*web3.Block
}

BlockEvent is an event emitted when a new block is included

type BlockTracker

type BlockTracker interface {
	Track(context.Context, func(block *web3.Block) error) error
}

BlockTracker is an interface to track new blocks on the chain

type Config

type Config struct {
	BatchSize          uint64
	MaxBlockBacklog    uint64
	EtherscanFastTrack bool
	EtherscanAPIKey    string
}

Config is the configuration of the tracker

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig returns the default tracker config

type Event

type Event struct {
	Type    EventType
	Added   []*web3.Log
	Removed []*web3.Log
}

Event is an event emitted when a new log is included

type EventType

type EventType int

EventType is the type of the event

const (
	// EventAdd happens when a new event is included in the chain
	EventAdd EventType = iota
	// EventDel may happen when there is a reorg and a past event is deleted
	EventDel
)

type Filter

type Filter struct {
	SyncCh  chan uint64
	EventCh chan *Event
	DoneCh  chan struct{}
	// contains filtered or unexported fields
}

Filter is a specific filter

func (*Filter) Entry

func (f *Filter) Entry() store.Entry

func (*Filter) GetLastBlock

func (f *Filter) GetLastBlock() (*web3.Block, error)

GetLastBlock returns the last block processed for this filter

func (*Filter) IsSynced

func (f *Filter) IsSynced() bool

IsSynced returns true if the filter is synced to head

func (*Filter) Sync

func (f *Filter) Sync(ctx context.Context) error

Sync syncs the filter

func (*Filter) SyncAsync

func (f *Filter) SyncAsync(ctx context.Context)

SyncAsync syncs the filter asynchronously

func (*Filter) Wait

func (f *Filter) Wait()

Wait waits the filter to finish

func (*Filter) WaitDuration

func (f *Filter) WaitDuration(dur time.Duration) error

WaitDuration waits for the filter to finish up to duration

type FilterConfig

type FilterConfig struct {
	Address []web3.Address `json:"address"`
	Topics  []*web3.Hash   `json:"topics"`
	Start   uint64
	Hash    string
	Async   bool
}

FilterConfig is a tracker filter configuration

type JSONBlockTracker

type JSONBlockTracker struct {
	PollInterval time.Duration
	// contains filtered or unexported fields
}

JSONBlockTracker implements the BlockTracker interface using the http jsonrpc endpoint

func NewJSONBlockTracker

func NewJSONBlockTracker(logger *log.Logger, provider Provider) *JSONBlockTracker

NewJSONBlockTracker creates a new json block tracker

func (*JSONBlockTracker) Track

func (k *JSONBlockTracker) Track(ctx context.Context, handle func(block *web3.Block) error) error

Track implements the BlockTracker interface

type Provider

type Provider interface {
	BlockNumber() (uint64, error)
	GetBlockByHash(hash web3.Hash, full bool) (*web3.Block, error)
	GetBlockByNumber(i web3.BlockNumber, full bool) (*web3.Block, error)
	GetLogs(filter *web3.LogFilter) ([]*web3.Log, error)
	ChainID() (*big.Int, error)
}

Provider are the eth1x methods required by the tracker

type SubscriptionBlockTracker

type SubscriptionBlockTracker struct {
	// contains filtered or unexported fields
}

SubscriptionBlockTracker is an interface to track new blocks using the newHeads subscription endpoint

func NewSubscriptionBlockTracker

func NewSubscriptionBlockTracker(logger *log.Logger, client *jsonrpc.Client) (*SubscriptionBlockTracker, error)

NewSubscriptionBlockTracker creates a new block tracker using the subscription endpoint

func (*SubscriptionBlockTracker) Track

func (s *SubscriptionBlockTracker) Track(ctx context.Context, handle func(block *web3.Block) error) error

Track implements the BlockTracker interface

type Tracker

type Tracker struct {
	BlockCh chan *BlockEvent

	ReadyCh chan struct{}
	// contains filtered or unexported fields
}

Tracker is a contract event tracker

func NewTracker

func NewTracker(provider Provider, config *Config) *Tracker

NewTracker creates a new tracker

func (*Tracker) GetSavedFilters

func (t *Tracker) GetSavedFilters() ([]*FilterConfig, error)

GetSavedFilters returns the filters stored in the store

func (*Tracker) NewFilter

func (t *Tracker) NewFilter(config *FilterConfig) (*Filter, error)

NewFilter creates a new log filter

func (*Tracker) SetLogger

func (t *Tracker) SetLogger(logger *log.Logger)

SetLogger sets a logger

func (*Tracker) SetStore

func (t *Tracker) SetStore(store store.Store)

SetStore sets the store

func (*Tracker) Start

func (t *Tracker) Start(ctx context.Context) error

Start starts the syncing

func (*Tracker) Sync

func (t *Tracker) Sync(ctx context.Context, filter *Filter) error

Sync syncs a specific filter

func (*Tracker) SyncAsync

func (t *Tracker) SyncAsync(ctx context.Context, filter *Filter)

SyncAsync syncs a specific filter asynchronously