Merge pull request #67 from BuxOrg/siggi/monitor-block-headers-import

Added block headers import

action_blockheader.go

@@ -6,7 +6,7 @@ import (
 	"github.com/libsv/go-bc"
 )
 
-// RecordBlockheader will Save a block header into the Datastore
+// RecordBlockHeader will Save a block header into the Datastore
 //
 // hash is the hash of the block header
 // bh is the block header data

chainstate/filters/metanet_test.go

@@ -1,8 +1,9 @@
 package filters
 
 import (
-	"github.com/mrz1836/go-whatsonchain"
 	"testing"
+
+	"github.com/mrz1836/go-whatsonchain"
 )
 
 func TestMetanet(t *testing.T) {

chainstate/filters/planaria-b_test.go

@@ -1,8 +1,9 @@
 package filters
 
 import (
-	"github.com/mrz1836/go-whatsonchain"
 	"testing"
+
+	"github.com/mrz1836/go-whatsonchain"
 )
 
 func TestPlanariaB(t *testing.T) {

chainstate/monitor.go

@@ -27,7 +27,6 @@ type Monitor struct {
 	maxNumberOfDestinations      int
 	processMempoolOnConnect      bool
 	filterType                   string
-	regexList                    []string
 	handler                      MonitorHandler
 }
 
@@ -128,7 +127,7 @@ func (m *Monitor) SaveDestinations() bool {
 	return m.saveTransactionsDestinations
 }
 
-// LooadMonitoredDestinations gets where we want to add the monitored destiantions from the database into the processor
+// LoadMonitoredDestinations gets where we want to add the monitored destinations from the database into the processor
 func (m *Monitor) LoadMonitoredDestinations() bool {
 	return m.loadMOnitoredDestinations
 }
@@ -144,7 +143,7 @@ func (m *Monitor) Monitor(handler MonitorHandler) error {
 	if m.client == nil {
 		handler.SetMonitor(m)
 		m.handler = handler
-		m.logger.Info(context.Background(), "[MONITOR] Connecting to server: %s", m.centrifugeServer)
+		m.logger.Info(context.Background(), fmt.Sprintf("[MONITOR] Connecting to server: %s", m.centrifugeServer))
 		m.client = newCentrifugeClient(m.centrifugeServer, handler)
 		if m.token != "" {
 			m.client.SetToken(m.token)

chainstate/monitor_client.go

@@ -32,20 +32,21 @@ func (a *AgentClient) Disconnect() error {
 	return a.Client.Disconnect()
 }
 
+// SetToken set the client token
 func (a *AgentClient) SetToken(token string) {
 	a.Client.SetToken(token)
 }
 
-// TODO: Just rely on the agent for this data type
 // AddFilterMessage defines a new filter to be published from the client
+// todo Just rely on the agent for this data type
 type AddFilterMessage struct {
 	Timestamp int64  `json:"timestamp"`
 	Regex     string `json:"regex"`
 	Filter    string `json:"filter"`
 	Hash      string `json:"hash"`
 }
 
-// AddFilter adds a new filtero the agent
+// AddFilter adds a new filter to the agent
 func (a *AgentClient) AddFilter(regex, item string) (centrifuge.PublishResult, error) {
 	msg := AddFilterMessage{
 		Regex:     regex,

chainstate/monitor_handler.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/libsv/go-bc"
-
 	"github.com/mrz1836/go-whatsonchain"
 )
 

chainstate/processor_test.go

@@ -4,13 +4,12 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/stretchr/testify/require"
-
 	"github.com/BuxOrg/bux/utils"
 	"github.com/bitcoinschema/go-bitcoin/v2"
 	"github.com/centrifugal/centrifuge-go"
 	"github.com/libsv/go-bt/v2/bscript"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	boom "github.com/tylertreat/BoomFilters"
 )
 

chainstate/types.go

@@ -1,9 +1,19 @@
 package chainstate
 
+// TransactionType tx types
 type TransactionType string
 
+// Metanet type
 const Metanet TransactionType = "metanet"
+
+// PubKeyHash type
 const PubKeyHash TransactionType = "pubkeyhash"
+
+// PlanariaB type
 const PlanariaB TransactionType = "planaria-b"
+
+// PlanariaD type
 const PlanariaD TransactionType = "planaria-d"
+
+// RareCandyFrogCartel type
 const RareCandyFrogCartel TransactionType = "rarecandy-frogcartel"

client.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/tonicpow/go-paymail"
-
 	"github.com/BuxOrg/bux/cachestore"
 	"github.com/BuxOrg/bux/chainstate"
 	"github.com/BuxOrg/bux/datastore"
@@ -14,6 +12,7 @@ import (
 	"github.com/BuxOrg/bux/taskmanager"
 	"github.com/BuxOrg/bux/utils"
 	"github.com/newrelic/go-agent/v3/newrelic"
+	"github.com/tonicpow/go-paymail"
 	"github.com/tonicpow/go-paymail/server"
 )
 
@@ -31,6 +30,7 @@ type (
 		dataStore     *dataStoreOptions     // Configuration options for the DataStore (MySQL, etc.)
 		debug         bool                  // If the client is in debug mode
 		encryptionKey string                // Encryption key for encrypting sensitive information (IE: paymail xPub) (hex encoded key)
+		httpClient    HTTPInterface         // HTTP interface to use
 		itc           bool                  // (Incoming Transactions Check) True will check incoming transactions via Miners (real-world)
 		iuc           bool                  // (Input UTXO Check) True will check input utxos when saving transactions
 		logger        logger.Interface      // Internal logging
@@ -172,7 +172,7 @@ func NewClient(ctx context.Context, opts ...ClientOps) (ClientInterface, error)
 
 	// Load the blockchain monitor
 	if client.options.chainstate.Monitor() != nil {
-		if err := client.loadMonitor(ctx); err != nil {
+		if err = client.loadMonitor(ctx); err != nil {
 			return nil, err
 		}
 	}
@@ -286,6 +286,11 @@ func (c *Client) Datastore() datastore.ClientInterface {
 	return nil
 }
 
+// HTTPClient will return the http interface to use in the client
+func (c *Client) HTTPClient() HTTPInterface {
+	return c.options.httpClient
+}
+
 // Logger will return the Logger if it exists
 func (c *Client) Logger() logger.Interface {
 	return c.options.logger

client_internal.go

@@ -109,6 +109,7 @@ func (c *Client) loadMonitor(ctx context.Context) (err error) {
 func (c *Client) runModelMigrations(models ...interface{}) (err error) {
 	d := c.Datastore()
 	for _, model := range models {
+		model.(ModelInterface).SetOptions(WithClient(c))
 		if err = model.(ModelInterface).Migrate(d); err != nil {
 			return
 		}

client_options.go

@@ -3,6 +3,7 @@ package bux
 import (
 	"context"
 	"database/sql"
+	"net/http"
 	"strings"
 	"time"
 
@@ -25,6 +26,10 @@ import (
 // ClientOps allow functional options to be supplied that overwrite default client options.
 type ClientOps func(c *clientOptions)
 
+const (
+	defaultHTTPTimeout = 20 * time.Second
+)
+
 // defaultClientOptions will return an clientOptions struct with the default settings
 //
 // Useful for starting with the default and then modifying as needed
@@ -59,6 +64,10 @@ func defaultClientOptions() *clientOptions {
 			options:         []datastore.ClientOps{},
 		},
 
+		httpClient: &http.Client{
+			Timeout: defaultHTTPTimeout,
+		},
+
 		// Blank model options (use the Base models)
 		models: &modelOptions{
 			modelNames:        modelNames(BaseModels...),
@@ -250,6 +259,20 @@ func WithIUCDisabled() ClientOps {
 	}
 }
 
+// WithImportBlockHeaders will import block headers on startup
+func WithImportBlockHeaders(blockHeadersZipFileURL string) ClientOps {
+	return func(c *clientOptions) {
+		c.dataStore.options = append(c.dataStore.options, datastore.WithImportBlockHeaders(blockHeadersZipFileURL))
+	}
+}
+
+// WithHTTPClient will set the custom http interface
+func WithHTTPClient(httpClient HTTPInterface) ClientOps {
+	return func(c *clientOptions) {
+		c.httpClient = httpClient
+	}
+}
+
 // WithLogger will set the custom logger interface
 func WithLogger(customLogger logger.Interface) ClientOps {
 	return func(c *clientOptions) {

datastore/client.go

@@ -20,20 +20,21 @@ type (
 
 	// clientOptions holds all the configuration for the client
 	clientOptions struct {
-		autoMigrate     bool              // Setting for Auto Migration of SQL tables
-		db              *gorm.DB          // Database connection for Read-Only requests (can be same as Write)
-		debug           bool              // Setting for global debugging
-		engine          Engine            // Datastore engine (MySQL, PostgreSQL, SQLite)
-		logger          logger.Interface  // Custom logger interface (from BUX)
-		loggerDB        glogger.Interface // Custom logger interface (for GORM)
-		migratedModels  []string          // List of models (types) that have been migrated
-		migrateModels   []interface{}     // Models for migrations
-		mongoDB         *mongo.Database   // Database connection for a MongoDB datastore
-		mongoDBConfig   *MongoDBConfig    // Configuration for a MongoDB datastore
-		newRelicEnabled bool              // If NewRelic is enabled (parent application)
-		sqlConfigs      []*SQLConfig      // Configuration for a MySQL or PostgreSQL datastore
-		sqLite          *SQLiteConfig     // Configuration for a SQLite datastore
-		tablePrefix     string            // Model table prefix
+		autoMigrate        bool              // Setting for Auto Migration of SQL tables
+		db                 *gorm.DB          // Database connection for Read-Only requests (can be same as Write)
+		debug              bool              // Setting for global debugging
+		engine             Engine            // Datastore engine (MySQL, PostgreSQL, SQLite)
+		importBlockHeaders string            // The URL of the block headers zip file to import old block headers on startup. if block 0 is found in the DB, block headers will mpt be downloaded
+		logger             logger.Interface  // Custom logger interface (from BUX)
+		loggerDB           glogger.Interface // Custom logger interface (for GORM)
+		migratedModels     []string          // List of models (types) that have been migrated
+		migrateModels      []interface{}     // Models for migrations
+		mongoDB            *mongo.Database   // Database connection for a MongoDB datastore
+		mongoDBConfig      *MongoDBConfig    // Configuration for a MongoDB datastore
+		newRelicEnabled    bool              // If NewRelic is enabled (parent application)
+		sqlConfigs         []*SQLConfig      // Configuration for a MySQL or PostgreSQL datastore
+		sqLite             *SQLiteConfig     // Configuration for a SQLite datastore
+		tablePrefix        string            // Model table prefix
 	}
 )
 

datastore/client_options.go

@@ -204,3 +204,10 @@ func WithLogger(customLogger logger.Interface) ClientOps {
 		}
 	}
 }
+
+// WithImportBlockHeaders will set the import block headers option
+func WithImportBlockHeaders(blockHeadersZipFileURL string) ClientOps {
+	return func(c *clientOptions) {
+		c.importBlockHeaders = blockHeadersZipFileURL
+	}
+}

datastore/interface.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"time"
 
+	"go.mongodb.org/mongo-driver/mongo"
 	"gorm.io/gorm"
 )
 
 // StorageService is the storage related methods
 type StorageService interface {
 	AutoMigrateDatabase(ctx context.Context, models ...interface{}) error
+	CreateInBatches(ctx context.Context, models interface{}, batchSize int) error
 	Execute(query string) *gorm.DB
 	GetModel(ctx context.Context, model interface{}, conditions map[string]interface{}, timeout time.Duration) error
 	GetModels(ctx context.Context, models interface{}, conditions map[string]interface{}, pageSize, page int,
@@ -33,7 +35,10 @@ type ClientInterface interface {
 	Engine() Engine
 	GetDatabaseName() string
 	GetTableName(modelName string) string
+	GetMongoCollection(collectionName string) *mongo.Collection
+	GetMongoCollectionByTableName(tableName string) *mongo.Collection
 	IsAutoMigrate() bool
 	IsDebug() bool
 	IsNewRelicEnabled() bool
+	ImportBlockHeadersFromURL() string
 }

datastore/migrate.go

@@ -60,6 +60,11 @@ func (c *Client) IsAutoMigrate() bool {
 	return c.options.autoMigrate
 }
 
+// ImportBlockHeadersFromURL will the URL where to import block headers from
+func (c *Client) ImportBlockHeadersFromURL() string {
+	return c.options.importBlockHeaders
+}
+
 // autoMigrateMongoDatabase will start a new database for Mongo
 func autoMigrateMongoDatabase(ctx context.Context, _ Engine, options *clientOptions,
 	_ ...interface{}) error {

datastore/models.go

@@ -127,6 +127,20 @@ func (c *Client) IncrementModel(
 	return
 }
 
+// CreateInBatches create all the models given in batches
+func (c *Client) CreateInBatches(
+	ctx context.Context,
+	models interface{},
+	batchSize int,
+) error {
+	if c.Engine() == MongoDB {
+		return c.CreateInBatchesMongo(ctx, models, batchSize)
+	}
+
+	tx := c.options.db.CreateInBatches(models, batchSize)
+	return tx.Error
+}
+
 // convertToInt64 will convert an interface to an int64
 func convertToInt64(i interface{}) int64 {
 	switch v := i.(type) {