Mx 16271 indexing incoming tokens

client/disabled/elasticClient.go

@@ -0,0 +1,73 @@
+package disabled
+
+import (
+	"bytes"
+	"context"
+)
+
+type elasticClient struct{}
+
+// NewDisabledElasticClient -
+func NewDisabledElasticClient() *elasticClient {
+	return &elasticClient{}
+}
+
+// DoBulkRequest -
+func (ec *elasticClient) DoBulkRequest(_ context.Context, _ *bytes.Buffer, _ string) error {
+	return nil
+}
+
+// DoQueryRemove -
+func (ec *elasticClient) DoQueryRemove(_ context.Context, _ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// DoMultiGet -
+func (ec *elasticClient) DoMultiGet(_ context.Context, _ []string, _ string, _ bool, _ interface{}) error {
+	return nil
+}
+
+// DoScrollRequest -
+func (ec *elasticClient) DoScrollRequest(_ context.Context, _ string, _ []byte, _ bool, _ func(responseBytes []byte) error) error {
+	return nil
+}
+
+// DoCountRequest -
+func (ec *elasticClient) DoCountRequest(_ context.Context, _ string, _ []byte) (uint64, error) {
+	return 0, nil
+}
+
+// UpdateByQuery -
+func (ec *elasticClient) UpdateByQuery(_ context.Context, _ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// PutMappings -
+func (ec *elasticClient) PutMappings(_ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// CheckAndCreateIndex -
+func (ec *elasticClient) CheckAndCreateIndex(_ string) error {
+	return nil
+}
+
+// CheckAndCreateAlias -
+func (ec *elasticClient) CheckAndCreateAlias(_ string, _ string) error {
+	return nil
+}
+
+// CheckAndCreateTemplate -
+func (ec *elasticClient) CheckAndCreateTemplate(_ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// CheckAndCreatePolicy -
+func (ec *elasticClient) CheckAndCreatePolicy(_ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// IsInterfaceNil - returns true if there is no value under the interface
+func (ec *elasticClient) IsInterfaceNil() bool {
+	return ec == nil
+}

client/disabled/elasticClient_test.go

@@ -0,0 +1,31 @@
+package disabled
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/multiversx/mx-chain-core-go/core/check"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDisabledElasticClient_MethodsShouldNotPanic(t *testing.T) {
+	t.Parallel()
+
+	ec := NewDisabledElasticClient()
+	require.False(t, check.IfNil(ec))
+
+	require.NotPanics(t, func() {
+		_ = ec.DoBulkRequest(context.Background(), new(bytes.Buffer), "")
+		_ = ec.DoQueryRemove(context.Background(), "", new(bytes.Buffer))
+		_ = ec.DoMultiGet(context.Background(), make([]string, 0), "", true, nil)
+		_ = ec.DoScrollRequest(context.Background(), "", []byte(""), true, nil)
+		_, _ = ec.DoCountRequest(context.Background(), "", []byte(""))
+		_ = ec.UpdateByQuery(context.Background(), "", new(bytes.Buffer))
+		_ = ec.PutMappings("", new(bytes.Buffer))
+		_ = ec.CheckAndCreateIndex("")
+		_ = ec.CheckAndCreateAlias("", "")
+		_ = ec.CheckAndCreateTemplate("", new(bytes.Buffer))
+		_ = ec.CheckAndCreatePolicy("", new(bytes.Buffer))
+	})
+}

cmd/elasticindexer/config/prefs.toml

@@ -23,3 +23,9 @@
         username = ""
         password = ""
         bulk-request-max-size-in-bytes = 4194304 # 4MB
+
+    [config.main-elastic-cluster]
+        enabled = true
+        url = "http://localhost:9201"
+        username = ""
+        password = ""

config/config.go

@@ -29,7 +29,8 @@ type Config struct {
 			LogsPath             string `toml:"logs-path"`
 		} `toml:"logs"`
 	} `toml:"config"`
-	Sovereign bool
+	Sovereign  bool
+	ESDTPrefix string
 }
 
 // ClusterConfig will hold the config for the Elasticsearch cluster
@@ -52,6 +53,12 @@ type ClusterConfig struct {
 			Password                  string `toml:"password"`
 			BulkRequestMaxSizeInBytes int    `toml:"bulk-request-max-size-in-bytes"`
 		} `toml:"elastic-cluster"`
+		MainChainCluster struct {
+			Enabled  bool   `toml:"enabled"`
+			URL      string `toml:"url"`
+			UserName string `toml:"username"`
+			Password string `toml:"password"`
+		} `toml:"main-elastic-cluster"`
 	} `toml:"config"`
 }
 

data/tokens.go

@@ -45,6 +45,18 @@ type SourceToken struct {
 	CurrentOwner string `json:"currentOwner"`
 }
 
+// ResponseTokenInfo is the structure for the tokens info response
+type ResponseTokenInfo struct {
+	Docs []ResponseTokenInfoDB `json:"docs"`
+}
+
+// ResponseTokenInfoDB is the structure for the token info response
+type ResponseTokenInfoDB struct {
+	Found  bool      `json:"found"`
+	ID     string    `json:"_id"`
+	Source TokenInfo `json:"_source"`
+}
+
 // TokenInfo is a structure that is needed to store information about a token
 type TokenInfo struct {
 	Name              string           `json:"name,omitempty"`

docker-compose.yml

@@ -15,6 +15,22 @@ services:
     ports:
       - "9200:9200"
       - "9300:9300"
+  main-elasticsearch:
+    container_name: es-container2
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.16.1
+    environment:
+      - "discovery.type=single-node"
+      - "xpack.security.enabled=false"
+      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    networks:
+      - es-net
+    ports:
+      - "9201:9200"
+      - "9301:9300"
   kibana:
     container_name: kb-container
     image: docker.elastic.co/kibana/kibana:7.16.1

factory/runType/interface.go

@@ -1,12 +1,13 @@
 package runType
 
 import (
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/transactions"
 )
 
 // RunTypeComponentsCreator is the interface for creating run type components
 type RunTypeComponentsCreator interface {
-	Create() *runTypeComponents
+	Create() (*runTypeComponents, error)
 	IsInterfaceNil() bool
 }
 
@@ -28,6 +29,7 @@ type RunTypeComponentsHandler interface {
 type RunTypeComponentsHolder interface {
 	TxHashExtractorCreator() transactions.TxHashExtractor
 	RewardTxDataCreator() transactions.RewardTxDataHandler
+	IndexTokensHandlerCreator() elasticproc.IndexTokensHandler
 	Create() error
 	Close() error
 	CheckSubcomponents() error

factory/runType/runTypeComponents.go

@@ -1,12 +1,14 @@
 package runType
 
 import (
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/transactions"
 )
 
 type runTypeComponents struct {
-	txHashExtractor transactions.TxHashExtractor
-	rewardTxData    transactions.RewardTxDataHandler
+	txHashExtractor    transactions.TxHashExtractor
+	rewardTxData       transactions.RewardTxDataHandler
+	indexTokensHandler elasticproc.IndexTokensHandler
 }
 
 // Close does nothing

factory/runType/runTypeComponentsFactory.go

@@ -1,6 +1,7 @@
 package runType
 
 import (
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/tokens"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/transactions"
 )
 
@@ -12,11 +13,12 @@ func NewRunTypeComponentsFactory() *runTypeComponentsFactory {
 }
 
 // Create will create the run type components
-func (rtcf *runTypeComponentsFactory) Create() *runTypeComponents {
+func (rtcf *runTypeComponentsFactory) Create() (*runTypeComponents, error) {
 	return &runTypeComponents{
-		txHashExtractor: transactions.NewTxHashExtractor(),
-		rewardTxData:    transactions.NewRewardTxData(),
-	}
+		txHashExtractor:    transactions.NewTxHashExtractor(),
+		rewardTxData:       transactions.NewRewardTxData(),
+		indexTokensHandler: tokens.NewDisabledIndexTokensHandler(),
+	}, nil
 }
 
 // IsInterfaceNil returns true if there is no value under the interface

factory/runType/runTypeComponentsHandler.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
 
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/transactions"
 )
 
@@ -34,7 +35,10 @@ func NewManagedRunTypeComponents(rtc RunTypeComponentsCreator) (*managedRunTypeC
 
 // Create will create the managed components
 func (mrtc *managedRunTypeComponents) Create() error {
-	rtc := mrtc.factory.Create()
+	rtc, err := mrtc.factory.Create()
+	if err != nil {
+		return err
+	}
 
 	mrtc.mutRunTypeCoreComponents.Lock()
 	mrtc.runTypeComponents = rtc
@@ -75,6 +79,9 @@ func (mrtc *managedRunTypeComponents) CheckSubcomponents() error {
 	if check.IfNil(mrtc.rewardTxData) {
 		return transactions.ErrNilRewardTxDataHandler
 	}
+	if check.IfNil(mrtc.indexTokensHandler) {
+		return transactions.ErrNilIndexTokensHandler
+	}
 	return nil
 }
 
@@ -102,6 +109,18 @@ func (mrtc *managedRunTypeComponents) RewardTxDataCreator() transactions.RewardT
 	return mrtc.runTypeComponents.rewardTxData
 }
 
+// IndexTokensHandlerCreator returns the index tokens handler
+func (mrtc *managedRunTypeComponents) IndexTokensHandlerCreator() elasticproc.IndexTokensHandler {
+	mrtc.mutRunTypeCoreComponents.Lock()
+	defer mrtc.mutRunTypeCoreComponents.Unlock()
+
+	if check.IfNil(mrtc.runTypeComponents) {
+		return nil
+	}
+
+	return mrtc.runTypeComponents.indexTokensHandler
+}
+
 // IsInterfaceNil returns true if the interface is nil
 func (mrtc *managedRunTypeComponents) IsInterfaceNil() bool {
 	return mrtc == nil

factory/runType/runTypeComponents_test.go

@@ -21,8 +21,9 @@ func TestRunTypeComponentsFactory_Create(t *testing.T) {
 	rtcf := NewRunTypeComponentsFactory()
 	require.NotNil(t, rtcf)
 
-	rtc := rtcf.Create()
+	rtc, err := rtcf.Create()
 	require.NotNil(t, rtc)
+	require.NoError(t, err)
 }
 
 func TestRunTypeComponentsFactory_Close(t *testing.T) {
@@ -31,8 +32,9 @@ func TestRunTypeComponentsFactory_Close(t *testing.T) {
 	rtcf := NewRunTypeComponentsFactory()
 	require.NotNil(t, rtcf)
 
-	rtc := rtcf.Create()
+	rtc, err := rtcf.Create()
 	require.NotNil(t, rtc)
+	require.NoError(t, err)
 
 	require.NoError(t, rtc.Close())
 }

factory/runType/sovereignRunTypeComponentsFactory.go

@@ -1,24 +1,73 @@
 package runType
 
 import (
+	"math"
+	"net/http"
+	"time"
+
+	"github.com/elastic/go-elasticsearch/v7"
+
+	"github.com/multiversx/mx-chain-es-indexer-go/client"
+	"github.com/multiversx/mx-chain-es-indexer-go/client/disabled"
+	"github.com/multiversx/mx-chain-es-indexer-go/client/logging"
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc"
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/factory"
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/tokens"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/transactions"
 )
 
-type sovereignRunTypeComponentsFactory struct{}
+type sovereignRunTypeComponentsFactory struct {
+	mainChainElastic factory.ElasticConfig
+	esdtPrefix       string
+}
 
 // NewSovereignRunTypeComponentsFactory will return a new instance of sovereign run type components factory
-func NewSovereignRunTypeComponentsFactory() *sovereignRunTypeComponentsFactory {
-	return &sovereignRunTypeComponentsFactory{}
+func NewSovereignRunTypeComponentsFactory(mainChainElastic factory.ElasticConfig, esdtPrefix string) *sovereignRunTypeComponentsFactory {
+	return &sovereignRunTypeComponentsFactory{
+		mainChainElastic: mainChainElastic,
+		esdtPrefix:       esdtPrefix,
+	}
 }
 
 // Create will create the run type components
-func (srtcf *sovereignRunTypeComponentsFactory) Create() *runTypeComponents {
+func (srtcf *sovereignRunTypeComponentsFactory) Create() (*runTypeComponents, error) {
+	mainChainElasticClient, err := createMainChainElasticClient(srtcf.mainChainElastic)
+	if err != nil {
+		return nil, err
+	}
+
+	sovIndexTokensHandler, err := tokens.NewSovereignIndexTokensHandler(srtcf.mainChainElastic.Enabled, mainChainElasticClient, srtcf.esdtPrefix)
+	if err != nil {
+		return nil, err
+	}
+
 	return &runTypeComponents{
-		txHashExtractor: transactions.NewSovereignTxHashExtractor(),
-		rewardTxData:    transactions.NewSovereignRewardTxData(),
+		txHashExtractor:    transactions.NewSovereignTxHashExtractor(),
+		rewardTxData:       transactions.NewSovereignRewardTxData(),
+		indexTokensHandler: sovIndexTokensHandler,
+	}, nil
+}
+
+func createMainChainElasticClient(mainChainElastic factory.ElasticConfig) (elasticproc.DatabaseClientHandler, error) {
+	if mainChainElastic.Enabled {
+		argsEsClient := elasticsearch.Config{
+			Addresses:     []string{mainChainElastic.Url},
+			Username:      mainChainElastic.UserName,
+			Password:      mainChainElastic.Password,
+			Logger:        &logging.CustomLogger{},
+			RetryOnStatus: []int{http.StatusConflict},
+			RetryBackoff:  retryBackOff,
+		}
+		return client.NewElasticClient(argsEsClient)
+	} else {
+		return disabled.NewDisabledElasticClient(), nil
 	}
 }
 
+func retryBackOff(attempt int) time.Duration {
+	return time.Duration(math.Exp2(float64(attempt))) * time.Second
+}
+
 // IsInterfaceNil returns true if there is no value under the interface
 func (srtcf *sovereignRunTypeComponentsFactory) IsInterfaceNil() bool {
 	return srtcf == nil