feat: add badger blocks db

internal/batcher/batcher.go

@@ -1,6 +0,0 @@
-package batcher
-
-type (
-	Batcher struct {
-	}
-)

internal/chain/block.go

@@ -39,3 +39,17 @@ func (c *Chain) GetBlock(ctx context.Context, blockNumber uint64) (types.Block,
 
 	return block, nil
 }
+
+func (c *Chain) GetLatestBlock(ctx context.Context) (uint64, error) {
+	var (
+		latestBlock big.Int
+	)
+
+	latestBlockCall := eth.BlockNumber().Returns(&latestBlock)
+
+	if err := c.provider.Client.CallCtx(ctx, latestBlockCall); err != nil {
+		return 0, err
+	}
+
+	return latestBlock.Uint64(), nil
+}

internal/chain/chain.go

@@ -19,7 +19,7 @@ type (
 	}
 )
 
-func NewChainProvider(o ChainOpts) (*Chain, error) {
+func New(o ChainOpts) (*Chain, error) {
 	providerOpts := celoutils.ProviderOpts{
 		RpcEndpoint: o.RPCEndpoint,
 		ChainId:     celoutils.MainnetChainId,

internal/db/blocks.go

@@ -0,0 +1,125 @@
+package db
+
+import (
+	"bytes"
+
+	"github.com/bits-and-blooms/bitset"
+	"github.com/dgraph-io/badger/v4"
+)
+
+func (d *DB) SetLowerBound(v uint64) error {
+	return d.setUint64(lowerBoundKey, v)
+}
+
+func (d *DB) GetLowerBound() (uint64, error) {
+	v, err := d.get(lowerBoundKey)
+	if err != nil {
+		return 0, err
+	}
+	return unmarshalUint64(v), nil
+}
+
+func (d *DB) SetUpperBound(v uint64) error {
+	return d.setUint64(upperBoundKey, v)
+}
+
+func (d *DB) GetUpperBound() (uint64, error) {
+	v, err := d.get(upperBoundKey)
+	if err != nil {
+		return 0, err
+	}
+	return unmarshalUint64(v), nil
+}
+
+func (d *DB) SetValue(v uint64) error {
+	return d.setUint64AsKey(v)
+}
+
+func (d *DB) GetMissingValuesBitSet(lowerBound uint64, upperBound uint64) (*bitset.BitSet, error) {
+	var (
+		b bitset.BitSet
+	)
+
+	err := d.db.View(func(txn *badger.Txn) error {
+		var (
+			lowerRaw = marshalUint64(lowerBound)
+			upperRaw = marshalUint64(upperBound)
+		)
+
+		for i := lowerBound; i <= upperBound; i++ {
+			b.Set(uint(i))
+		}
+
+		opts := badger.DefaultIteratorOptions
+		opts.PrefetchValues = false
+
+		iter := txn.NewIterator(opts)
+		defer iter.Close()
+
+		for iter.Seek(lowerRaw); iter.Valid(); iter.Next() {
+			k := iter.Item().Key()
+
+			if bytes.Compare(k, upperRaw) > 0 {
+				return nil
+			}
+
+			b.Clear(uint(unmarshalUint64(k)))
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &b, nil
+}
+
+func (d *DB) Cleanup() error {
+	var (
+		safeToDeleteKeys [][]byte
+	)
+
+	err := d.db.View(func(txn *badger.Txn) error {
+		lowerBound, err := d.get(lowerBoundKey)
+		if err != nil {
+			return err
+		}
+
+		lowerBound = marshalUint64(unmarshalUint64(lowerBound) - 1)
+
+		opts := badger.DefaultIteratorOptions
+		opts.PrefetchValues = false
+
+		it := txn.NewIterator(opts)
+		defer it.Close()
+
+		for it.Rewind(); it.Valid(); it.Next() {
+			k := it.Item().Key()
+
+			if bytes.Compare(k, lowerBound) > 0 {
+				return nil
+			}
+
+			safeToDeleteKeys = append(safeToDeleteKeys, it.Item().KeyCopy(nil))
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	wb := d.db.NewWriteBatch()
+	for _, k := range safeToDeleteKeys {
+		if err := wb.Delete(k); err != nil {
+			return nil
+		}
+	}
+
+	if err := wb.Flush(); err != nil {
+		return err
+	}
+
+	return nil
+}

internal/db/db.go

@@ -0,0 +1,96 @@
+package db
+
+import (
+	"encoding/binary"
+	"log/slog"
+
+	"github.com/dgraph-io/badger/v4"
+)
+
+type (
+	DBOpts struct {
+		Logg *slog.Logger
+	}
+
+	DB struct {
+		db   *badger.DB
+		logg *slog.Logger
+	}
+)
+
+const (
+	dbFolderName = "celo_tracker_blocks_db"
+
+	upperBoundKey = "upper"
+	lowerBoundKey = "lower"
+)
+
+var (
+	sortableOrder = binary.BigEndian
+)
+
+func New(o DBOpts) (*DB, error) {
+	opts := badger.DefaultOptions(dbFolderName)
+	opts.Logger = nil
+	db, err := badger.Open(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	return &DB{
+		db:   db,
+		logg: o.Logg,
+	}, nil
+}
+
+func (d *DB) Close() error {
+	return d.db.Close()
+}
+
+func (d *DB) get(k string) ([]byte, error) {
+	var v []byte
+	err := d.db.View(func(txn *badger.Txn) error {
+		item, err := txn.Get([]byte(k))
+		if err != nil {
+			return err
+		}
+		v, err = item.ValueCopy(nil)
+		return err
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return v, nil
+}
+
+func (d *DB) setUint64(k string, v uint64) error {
+	err := d.db.Update(func(txn *badger.Txn) error {
+		return txn.Set([]byte(k), marshalUint64(v))
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (d *DB) setUint64AsKey(v uint64) error {
+	err := d.db.Update(func(txn *badger.Txn) error {
+		return txn.Set(marshalUint64(v), nil)
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func unmarshalUint64(b []byte) uint64 {
+	return sortableOrder.Uint64(b)
+}
+
+func marshalUint64(v uint64) []byte {
+	b := make([]byte, 8)
+	sortableOrder.PutUint64(b, v)
+	return b
+}

internal/processor/processor.go

@@ -5,12 +5,12 @@ import (
 	"log/slog"
 
 	"github.com/alitto/pond"
-	"github.com/celo-org/celo-blockchain/common"
 	"github.com/celo-org/celo-blockchain/core/types"
 	"github.com/ef-ds/deque/v2"
-	"github.com/grassrootseconomics/celo-events/internal/chain"
-	"github.com/grassrootseconomics/celo-events/internal/pool"
-	"github.com/grassrootseconomics/celo-events/internal/stats"
+	"github.com/grassrootseconomics/celo-tracker/internal/chain"
+	"github.com/grassrootseconomics/celo-tracker/internal/db"
+	"github.com/grassrootseconomics/celo-tracker/internal/pool"
+	"github.com/grassrootseconomics/celo-tracker/internal/stats"
 )
 
 type (
@@ -19,6 +19,7 @@ type (
 		BlocksQueue *deque.Deque[types.Block]
 		Logg        *slog.Logger
 		Stats       *stats.Stats
+		DB          *db.DB
 	}
 
 	Processor struct {
@@ -27,6 +28,7 @@ type (
 		blocksQueue *deque.Deque[types.Block]
 		logg        *slog.Logger
 		stats       *stats.Stats
+		db          *db.DB
 	}
 )
 
@@ -37,17 +39,19 @@ func NewProcessor(o ProcessorOpts) *Processor {
 		blocksQueue: o.BlocksQueue,
 		logg:        o.Logg,
 		stats:       o.Stats,
+		db:          o.DB,
 	}
 }
 
-func (p *Processor) Start(ctx context.Context) error {
+func (p *Processor) Start(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
 			p.logg.Info("block processor shutting down")
-			return nil
+			p.Stop()
+			return
 		default:
-			for p.blocksQueue.Len() > 0 {
+			if p.blocksQueue.Len() > 0 {
 				v, _ := p.blocksQueue.PopFront()
 				p.pool.Submit(func() {
 					if err := p.processBlock(v); err != nil {
@@ -65,8 +69,9 @@ func (p *Processor) Stop() {
 
 func (p *Processor) processBlock(block types.Block) error {
 	ctx := context.Background()
+	blockNumber := block.NumberU64()
 
-	transactionsResp, err := p.chain.GetTransactions(ctx, block)
+	_, err := p.chain.GetTransactions(ctx, block)
 	if err != nil {
 		return err
 	}
@@ -76,12 +81,16 @@ func (p *Processor) processBlock(block types.Block) error {
 		return err
 	}
 
-	for i, receipt := range receiptsResp {
+	for _, receipt := range receiptsResp {
 		if receipt.Status < 1 {
-			p.logg.Warn("reverted receipt", "tx_hash", receipt.TxHash, "input_data", common.Bytes2Hex(transactionsResp[i].Data()))
+			//
 		}
-		p.logg.Info("successful receipt", "tx_hash", receipt.TxHash, "status", receipt.Status)
 	}
 
+	if err := p.db.SetValue(blockNumber); err != nil {
+		return err
+	}
+	p.logg.Debug("successfully processed block", "block", blockNumber)
+
 	return nil
 }

internal/queue/queue.go

@@ -1,13 +0,0 @@
-package queue
-
-import (
-	"github.com/celo-org/celo-blockchain/core/types"
-	"github.com/ef-ds/deque/v2"
-)
-
-type (
-	Queue struct {
-		BlocksQueue *deque.Deque[types.Block]
-		BatchQueue  *deque.Deque[uint64]
-	}
-)

internal/stats/stats.go

@@ -22,7 +22,7 @@ type (
 	}
 )
 
-func NewStats(logg *slog.Logger) *Stats {
+func New(logg *slog.Logger) *Stats {
 	return &Stats{
 		logg: logg,
 	}

internal/syncer/historical.go

@@ -3,42 +3,71 @@ package syncer
 import (
 	"context"
 	"fmt"
+
+	"github.com/dgraph-io/badger/v4"
 )
 
-func (s *Syncer) BootstrapHistoricalSyncer() {
-	// logg here
-	for i, e := s.db.NextSet(0); e; i, e = s.db.NextSet(i + 1) {
-		if i > 0 {
-			s.batchQueue.PushBack(uint64(i))
+const (
+	blockBatchSize = 100
+)
+
+func (s *Syncer) BootstrapHistoricalSyncer() error {
+	v, err := s.db.GetLowerBound()
+	if err != nil {
+		if err == badger.ErrKeyNotFound {
+			if err := s.db.SetLowerBound(s.initialLowerBound); err != nil {
+				return err
+			}
+			v = s.initialLowerBound
+		} else {
+			return err
 		}
 	}
+
+	latestBlock, err := s.chain.GetLatestBlock(context.Background())
+	if err != nil {
+		return err
+	}
+	if err := s.db.SetUpperBound(latestBlock); err != nil {
+		return err
+	}
+
+	missingBlocks, err := s.db.GetMissingValuesBitSet(v, latestBlock)
+	if err != nil {
+		return err
+	}
+	missingBlocksCount := missingBlocks.Count()
+	s.logg.Info("bootstrapping historical syncer", "missing_blocks", missingBlocksCount, "lower_bound", v, "upper_bound", latestBlock)
+
+	buffer := make([]uint, missingBlocksCount)
+	missingBlocks.NextSetMany(0, buffer)
+	for _, v := range buffer {
+		s.batchQueue.PushFront(uint64(v))
+	}
+
+	return nil
 }
 
 func (s *Syncer) StartHistoricalSyncer(ctx context.Context) error {
+	s.logg.Info("starting historical syncer", "batch_size", blockBatchSize)
 	for {
 		select {
 		case <-ctx.Done():
 			s.logg.Info("historical syncer shutting down")
 			return nil
 		default:
-			for s.batchQueue.Len() > 0 {
+			if s.batchQueue.Len() > 0 {
 				var (
 					currentIterLen = s.batchQueue.Len()
-					batch          []uint64
 				)
 
-				if currentIterLen < blockBatchSize {
-					batch = make([]uint64, currentIterLen)
-					for i := 0; i < currentIterLen; i++ {
-						v, _ := s.batchQueue.PopFront()
-						batch[i] = v
-					}
-				} else {
-					batch = make([]uint64, blockBatchSize)
-					for i := 0; i < blockBatchSize; i++ {
-						v, _ := s.batchQueue.PopFront()
-						batch[i] = v
-					}
+				if currentIterLen > blockBatchSize {
+					currentIterLen = blockBatchSize
+				}
+				batch := make([]uint64, currentIterLen)
+				for i := 0; i < currentIterLen; i++ {
+					v, _ := s.batchQueue.PopFront()
+					batch[i] = v
 				}
 
 				blocks, err := s.chain.GetBlocks(context.Background(), batch)

internal/syncer/syncer.go

@@ -1,18 +1,15 @@
 package syncer
 
 import (
+	"errors"
 	"log/slog"
 
-	"github.com/bits-and-blooms/bitset"
 	"github.com/celo-org/celo-blockchain/core/types"
 	"github.com/celo-org/celo-blockchain/ethclient"
 	"github.com/ef-ds/deque/v2"
-	"github.com/grassrootseconomics/celo-events/internal/chain"
-	"github.com/grassrootseconomics/celo-events/internal/stats"
-)
-
-const (
-	blockBatchSize = 100
+	"github.com/grassrootseconomics/celo-tracker/internal/chain"
+	"github.com/grassrootseconomics/celo-tracker/internal/db"
+	"github.com/grassrootseconomics/celo-tracker/internal/stats"
 )
 
 type (
@@ -23,34 +20,40 @@ type (
 		Chain             *chain.Chain
 		Logg              *slog.Logger
 		Stats             *stats.Stats
-		// replace with db
-		Db *bitset.BitSet
+		DB                *db.DB
+		InitialLowerBound uint64
 	}
 
 	Syncer struct {
-		batchQueue  *deque.Deque[uint64]
-		blocksQueue *deque.Deque[types.Block]
-		chain       *chain.Chain
-		logg        *slog.Logger
-		stats       *stats.Stats
-		ethClient   *ethclient.Client
-		db          *bitset.BitSet
+		batchQueue        *deque.Deque[uint64]
+		blocksQueue       *deque.Deque[types.Block]
+		chain             *chain.Chain
+		logg              *slog.Logger
+		stats             *stats.Stats
+		ethClient         *ethclient.Client
+		db                *db.DB
+		initialLowerBound uint64
 	}
 )
 
-func NewSyncer(o SyncerOpts) (*Syncer, error) {
+func New(o SyncerOpts) (*Syncer, error) {
+	if o.InitialLowerBound == 0 {
+		return nil, errors.New("initial lower bound not set")
+	}
+
 	ethClient, err := ethclient.Dial(o.WebSocketEndpoint)
 	if err != nil {
 		return nil, err
 	}
 
 	return &Syncer{
-		batchQueue:  o.BatchQueue,
-		blocksQueue: o.BlocksQueue,
-		chain:       o.Chain,
-		logg:        o.Logg,
-		stats:       o.Stats,
-		ethClient:   ethClient,
-		db:          o.Db,
+		batchQueue:        o.BatchQueue,
+		blocksQueue:       o.BlocksQueue,
+		chain:             o.Chain,
+		logg:              o.Logg,
+		stats:             o.Stats,
+		ethClient:         ethClient,
+		db:                o.DB,
+		initialLowerBound: o.InitialLowerBound,
 	}, nil
 }