feat: Node client

config/load_test.go

@@ -29,7 +29,7 @@ func Test_Load(t *testing.T) {
 		require.NoError(t, err, "error loading config")
 
 		// then
-		// verify not overriden default example value
+		// verify not overridden default example value
 		assert.Equal(t, expectedConfig.GrpcMessageSize, actualConfig.GrpcMessageSize)
 
 		// verify correct override

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/go-testfixtures/testfixtures/v3 v3.9.0
 	github.com/go-zeromq/zmq4 v0.17.0
 	github.com/golang-migrate/migrate/v4 v4.16.2
+	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0

internal/blocktx/integration_test/reorg_integration_test.go

@@ -25,17 +25,18 @@ import (
 	"testing"
 	"time"
 
-	"github.com/bitcoin-sv/arc/internal/blocktx"
-	"github.com/bitcoin-sv/arc/internal/blocktx/blocktx_api"
-	"github.com/bitcoin-sv/arc/internal/blocktx/store/postgresql"
-	testutils "github.com/bitcoin-sv/arc/internal/test_utils"
 	_ "github.com/golang-migrate/migrate/v4/source/file"
 	_ "github.com/lib/pq"
 	"github.com/libsv/go-p2p"
 	"github.com/libsv/go-p2p/chaincfg/chainhash"
 	"github.com/libsv/go-p2p/wire"
 	"github.com/ory/dockertest/v3"
 	"github.com/stretchr/testify/require"
+
+	"github.com/bitcoin-sv/arc/internal/blocktx"
+	"github.com/bitcoin-sv/arc/internal/blocktx/blocktx_api"
+	"github.com/bitcoin-sv/arc/internal/blocktx/store/postgresql"
+	testutils "github.com/bitcoin-sv/arc/internal/test_utils"
 )
 
 const (
@@ -118,7 +119,7 @@ func TestBlockStatus(t *testing.T) {
 	blockMessage := &p2p.BlockMessage{
 		Header: &wire.BlockHeader{
 			Version:    541065216,
-			PrevBlock:  *prevBlockHash, // NON-existant in the db
+			PrevBlock:  *prevBlockHash, // NON-existent in the db
 			MerkleRoot: *merkleRoot,
 			Bits:       0x1d00ffff,
 		},

internal/blocktx/processor.go

@@ -448,7 +448,7 @@ func (p *Processor) processBlock(msg *p2p.BlockMessage) error {
 		// with no blocks, to mark the first block as the LONGEST chain
 		longestTipExists, err = p.longestTipExists(ctx)
 		if err != nil {
-			p.logger.Error("unable to verify the longest tip existance in db", slog.String("hash", blockHash.String()), slog.Uint64("height", msg.Height), slog.String("err", err.Error()))
+			p.logger.Error("unable to verify the longest tip existence in db", slog.String("hash", blockHash.String()), slog.Uint64("height", msg.Height), slog.String("err", err.Error()))
 			return err
 		}
 	}

internal/node_client/config/bitcoin.conf

@@ -0,0 +1,145 @@
+server=1
+rest=1
+listen=1
+regtest=1
+printtoconsole=1
+txindex=1
+dnsseed=0
+upnp=0
+usecashaddr=0
+debug=1
+
+#Bind to given address to listen for JSON-RPC connections. Use
+#[host]:port notation for IPv6. This option can be specified
+#multiple times (default: bind to all interfaces)
+#rpcbind=<addr>
+
+rpcport=18332
+rpcuser=bitcoin
+rpcpassword=bitcoin
+rpcallowip=0.0.0.0/0
+
+port=18333
+
+# onlynet=ipv4
+listenonion=0
+# prune=550
+
+addressindex=1
+timestampindex=1
+spentindex=1
+
+#Allow JSON-RPC connections from specified source. Valid for <ip> are a
+#single IP (e.g. 1.2.3.4), a network/netmask (e.g.
+#1.2.3.4/255.255.255.0) or a network/CIDR (e.g. 1.2.3.4/24). This
+#option can be specified multiple times
+#rpcallowip=<ip>
+#Set the number of threads to service RPC calls (default: 4)
+rpcthreads=24
+
+#Set the depth of the work queue to service RPC calls (default: 16)
+rpcworkqueue=600
+
+#Timeout during HTTP requests (default: 30)
+#rpcservertimeout=<n>
+blockmaxsize=512000000
+excessiveblocksize=2000000000
+maxstackmemoryusageconsensus=200000000
+
+#connect=142.93.36.189:9988
+
+#node filter
+#whitelist=127.0.0.1
+
+#Append comment to the user agent string
+#uacomment=<cmt>
+
+#Maximum database write batch size in bytes (default: 16777216)
+#dbbatchsize=16777216
+
+#Set database cache size in megabytes (4 to 16384, default: 450)
+dbcache=16384
+
+#Limit size of signature cache to <n> MiB (default: 32)
+maxsigcachesize=260
+
+#Limit size of script cache to <n> MiB (default: 32)
+maxscriptcachesize=260
+
+#Keep at most <n> unconnectable transactions in memory (default: 100)
+maxorphantx=100000
+
+#Keep the transaction memory pool below <n> megabytes (default: 300)
+maxmempool=2000
+
+#Do not keep transactions in the mempool longer than <n> hours (default: 336)
+#mempoolexpiry=48
+
+#Extra transactions to keep in memory for compact block reconstructions(default: 100)
+blockreconstructionextratxn=100000
+
+#Set the number of script verification threads (-4 to 16, 0 = auto, <0 =
+#leave that many cores free, default: 0)
+#par=<n>
+
+#Threshold for disconnecting misbehaving peers (default: 100)
+banscore=10000
+
+#Number of seconds to keep misbehaving peers from reconnecting (default: 86400)
+#bantime=<n>
+
+#Maintain at most <n> connections to peers (default: 125)
+#maxconnections=<n>
+
+#Maximum per-connection receive buffer, <n>*1000 bytes (default: 5000)
+#maxreceivebuffer=<n>
+
+#Maximum per-connection send buffer, <n>*1000 bytes (default: 1000)
+#maxsendbuffer=<n>
+
+#Specify connection timeout in milliseconds (minimum: 1, default: 5000)
+#timeout=<n>
+
+#ZeroMQ notification options:
+
+#Enable publish hash block in <address>
+zmqpubhashblock=tcp://*:28332
+
+#Enable publish hash transaction in <address>
+zmqpubhashtx=tcp://*:28332
+zmqpubhashtx2=tcp://*:28332
+
+#Enable publish raw block in <address>
+zmqpubrawblock=tcp://*:28332
+zmqpubrawblock2=tcp://*:28332
+
+#Enable publish raw transaction in <address>
+zmqpubrawtx=tcp://*:28332
+
+invalidtxsink=ZMQ
+zmqpubinvalidtx=tcp://*:28332
+zmqpubdiscardedfrommempool=tcp://*:28332
+
+#Do not accept transactions if number of in-mempool ancestors is <n> or more (default: 1000)
+#limitancestorcount=<n>
+
+#Do not accept transactions whose size with all in-mempool ancestors exceeds <n> kilobytes (default: 101)
+#limitancestorsize=<n>
+
+#Do not accept transactions if any ancestor would have <n> or more in-mempool descendants (default: 25)
+#limitdescendantcount=<n>
+
+#Do not accept transactions if any ancestor would have more than <n> kilobytes of in-mempool descendants (default: 101).
+#limitdescendantsize=<n>
+
+#Fees (in BCH/kB) smaller than this are considered zero fee for relaying, mining and transaction creation (default: 0.00001)
+#minrelaytxfee=<amt>
+
+#Relay and mine "non-standard" transactions (testnet/regtest only; default: 1)
+#acceptnonstdtxn=0
+
+#Use genesis rules from the 1st block instead of waiting until 10,000 (default) blocks to activate genesis rules.
+genesisactivationheight=1
+
+minminingtxfee=0.0000005
+whitelist=172.20.0.1/32

internal/node_client/node_client.go

@@ -1,3 +1,111 @@
 package node_client
 
-// Todo: Implement node client
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"runtime"
+
+	sdkTx "github.com/bitcoin-sv/go-sdk/transaction"
+	"github.com/ordishs/go-bitcoin"
+	"go.opentelemetry.io/otel/attribute"
+
+	"github.com/bitcoin-sv/arc/internal/tracing"
+)
+
+var (
+	ErrFailedToGetRawTransaction   = errors.New("failed to get raw transaction")
+	ErrFailedToGetMempoolAncestors = errors.New("failed to get mempool ancestors")
+)
+
+type NodeClient struct {
+	bitcoinClient     *bitcoin.Bitcoind
+	tracingEnabled    bool
+	tracingAttributes []attribute.KeyValue
+}
+
+func WithTracer(attr ...attribute.KeyValue) func(s *NodeClient) {
+	return func(p *NodeClient) {
+		p.tracingEnabled = true
+		if len(attr) > 0 {
+			p.tracingAttributes = append(p.tracingAttributes, attr...)
+		}
+		_, file, _, ok := runtime.Caller(1)
+		if ok {
+			p.tracingAttributes = append(p.tracingAttributes, attribute.String("file", file))
+		}
+	}
+}
+
+func New(n *bitcoin.Bitcoind, opts ...func(client *NodeClient)) (NodeClient, error) {
+	node := NodeClient{
+		bitcoinClient: n,
+	}
+
+	for _, opt := range opts {
+		opt(&node)
+	}
+
+	return node, nil
+}
+
+func (n NodeClient) GetMempoolAncestors(ctx context.Context, ids []string) ([]string, error) {
+	_, span := tracing.StartTracing(ctx, "NodeClient_GetMempoolAncestors", n.tracingEnabled, n.tracingAttributes...)
+	defer tracing.EndTracing(span)
+
+	uniqueIDs := make(map[string]struct{})
+
+	for _, id := range ids {
+		_, span := tracing.StartTracing(ctx, "Bitcoind_GetMempoolAncestors", n.tracingEnabled, n.tracingAttributes...)
+		nTx, err := n.bitcoinClient.GetMempoolAncestors(id, false)
+		tracing.EndTracing(span)
+		if err != nil {
+			return nil, errors.Join(ErrFailedToGetMempoolAncestors, err)
+		}
+
+		if nTx == nil {
+			return nil, nil
+		}
+
+		var txIDs []string
+
+		err = json.Unmarshal(nTx, &txIDs)
+		if err != nil {
+			return nil, fmt.Errorf("failed to unmarshal raw transaction: %v", err)
+		}
+
+		for _, txID := range txIDs {
+			_, found := uniqueIDs[txID]
+			if !found {
+				uniqueIDs[txID] = struct{}{}
+			}
+		}
+	}
+
+	allTxIDs := make([]string, len(uniqueIDs))
+	counter := 0
+	for id := range uniqueIDs {
+		allTxIDs[counter] = id
+		counter++
+	}
+	return allTxIDs, nil
+}
+
+func (n NodeClient) GetRawTransaction(ctx context.Context, id string) (*sdkTx.Transaction, error) {
+	_, span := tracing.StartTracing(ctx, "NodeClient_GetRawTransaction", n.tracingEnabled, n.tracingAttributes...)
+	defer tracing.EndTracing(span)
+
+	nTx, err := n.bitcoinClient.GetRawTransaction(id)
+
+	if err != nil {
+		return nil, errors.Join(ErrFailedToGetRawTransaction, err)
+	}
+
+	rt, err := sdkTx.NewTransactionFromHex(nTx.Hex)
+	if err != nil {
+		return nil, err
+	}
+
+	return rt, nil
+}