Merge remote-tracking branch 'origin/feature/bitcoin-indexer' into feat/runes-module

core/datasources/bitcoin_node.go

@@ -3,6 +3,7 @@ package datasources
 import (
 	"context"
 
+	"github.com/cockroachdb/errors"
 	"github.com/gaze-network/indexer-network/core/indexers"
 	"github.com/gaze-network/indexer-network/core/types"
 )
@@ -10,13 +11,40 @@ import (
 // Make sure to implement the BitcoinDatasource interface
 var _ indexers.BitcoinDatasource = (*BitcoinNodeDatasource)(nil)
 
+// BitcoinNodeDatasource fetch data from Bitcoin node for Bitcoin Indexer
 type BitcoinNodeDatasource struct{}
 
+// Fetch polling blocks from Bitcoin node
+//
+//   - from: block height to start fetching, if -1, it will start from genesis block
+//   - to: block height to stop fetching, if -1, it will fetch until the latest block
 func (d *BitcoinNodeDatasource) Fetch(ctx context.Context, from, to int64) ([]*types.Block, error) {
-	return nil, nil
+	ch, stop, err := d.FetchAsync(ctx, from, to)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	blocks := make([]*types.Block, 0)
+	for {
+		select {
+		case b, ok := <-ch:
+			if !ok {
+				return blocks, nil
+			}
+			blocks = append(blocks, b...)
+		case <-ctx.Done():
+			stop()
+			return nil, errors.Wrap(ctx.Err(), "context done")
+		}
+	}
 }
 
-func (d *BitcoinNodeDatasource) FetchAsync(ctx context.Context, from, to int64) (chan<- []*types.Block, func(), error) {
+// FetchAsync polling blocks from Bitcoin node asynchronously (non-blocking)
+//
+//   - from: block height to start fetching, if -1, it will start from genesis block
+//   - to: block height to stop fetching, if -1, it will fetch until the latest block
+func (d *BitcoinNodeDatasource) FetchAsync(ctx context.Context, from, to int64) (<-chan []*types.Block, func(), error) {
 	ctx, cancel := context.WithCancel(ctx)
+	_ = ctx
 	return nil, cancel, nil
 }

core/indexers/bitcoin_indexer.go

@@ -22,7 +22,7 @@ type (
 // Make sure to implement the IndexerWorker interface
 var _ core.IndexerWorker = (*BitcoinIndexer)(nil)
 
-// BitcoinIndexer is the indexer for sync Bitcoin data to the database.
+// BitcoinIndexer is the polling indexer for sync Bitcoin data to the database.
 type BitcoinIndexer struct {
 	Processor    BitcoinProcessor
 	Datasource   BitcoinDatasource

core/indexers/interface.go

@@ -15,6 +15,9 @@ type Processor[T any] interface {
 	// CurrentBlock returns the latest indexed block header.
 	CurrentBlock(ctx context.Context) (types.BlockHeader, error)
 
+	// GetIndexedBlock returns the indexed block header by the specified block height.
+	GetIndexedBlock(ctx context.Context, height int64) (types.BlockHeader, error)
+
 	// RevertData revert synced data to the specified block height for re-indexing.
 	RevertData(ctx context.Context, from int64) error
 }

modules/bitcoin/processor.go

@@ -73,9 +73,8 @@ func (p *Processor) CurrentBlock(ctx context.Context) (types.BlockHeader, error)
 	return b, nil
 }
 
-func (p *Processor) PrepareData(ctx context.Context, from, to int64) ([]*types.Block, error) {
-	// TODO: move out to a separate interface (e.g. DataFetcher)
-	return nil, nil
+func (p *Processor) GetIndexedBlock(ctx context.Context, height int64) (types.BlockHeader, error) {
+	return types.BlockHeader{}, nil
 }
 
 func (p *Processor) RevertData(ctx context.Context, from int64) error {