Add register_global_constant

cmd/mempool/config/validators.go

@@ -41,7 +41,7 @@ func (cfg MempoolDataSource) Validate() error {
 func (cfg Filters) Validate() error {
 	switch {
 	case len(cfg.Kinds) == 0:
-		cfg.Kinds = append(cfg.Kinds, node.KindTransaction)
+		cfg.Kinds = []string{node.KindTransaction}
 	default:
 		if err := validateKinds(cfg.Kinds...); err != nil {
 			return err
@@ -62,7 +62,7 @@ func validateKinds(kinds ...string) error {
 		for _, valid := range []string{
 			node.KindActivation, node.KindBallot, node.KindDelegation, node.KindDoubleBaking, node.KindDoubleEndorsing,
 			node.KindEndorsement, node.KindNonceRevelation, node.KindOrigination, node.KindProposal,
-			node.KindReveal, node.KindTransaction,
+			node.KindReveal, node.KindTransaction, node.KindRegisterGlobalConstant,
 		} {
 			if kind == valid {
 				found = true

cmd/mempool/handlers.go

@@ -178,6 +178,8 @@ func (indexer *Indexer) handleContent(db *gorm.DB, content node.Content, operati
 		return handleReveal(db, content, operation, indexer.filters.Accounts...)
 	case node.KindTransaction:
 		return handleTransaction(db, content, operation, indexer.filters.Accounts...)
+	case node.KindRegisterGlobalConstant:
+		return handleRegisterGloabalConstant(db, content, operation)
 	default:
 		return errors.Wrap(node.ErrUnknownKind, content.Kind)
 	}
@@ -345,6 +347,15 @@ func handleProposal(db *gorm.DB, content node.Content, operation models.MempoolO
 	return nil
 }
 
+func handleRegisterGloabalConstant(db *gorm.DB, content node.Content, operation models.MempoolOperation) error {
+	var registerGlobalConstant models.RegisterGlobalConstant
+	if err := json.Unmarshal(content.Body, &registerGlobalConstant); err != nil {
+		return err
+	}
+	registerGlobalConstant.MempoolOperation = operation
+	return db.Clauses(clause.OnConflict{DoNothing: true}).Create(&registerGlobalConstant).Error
+}
+
 func (indexer *Indexer) isKindAvailiable(kind string) bool {
 	for _, availiable := range indexer.filters.Kinds {
 		if strings.HasPrefix(kind, availiable) {

cmd/mempool/models/mempool_operation.go

@@ -192,6 +192,8 @@ func getModelByKind(kind string) (interface{}, error) {
 		return &Reveal{}, nil
 	case node.KindTransaction:
 		return &Transaction{}, nil
+	case node.KindRegisterGlobalConstant:
+		return &RegisterGlobalConstant{}, nil
 	default:
 		return nil, errors.Wrap(node.ErrUnknownKind, kind)
 	}

cmd/mempool/models/register_global_constant.go

@@ -0,0 +1,19 @@
+package models
+
+import "gorm.io/datatypes"
+
+// RegisterGlobalConstant -
+type RegisterGlobalConstant struct {
+	MempoolOperation
+	Source       string         `json:"source"`
+	Fee          string         `json:"fee"`
+	Counter      string         `json:"counter"`
+	GasLimit     string         `json:"gas_limit"`
+	StorageLimit string         `json:"storage_limit"`
+	Value        datatypes.JSON `json:"value"`
+}
+
+// TableName -
+func (RegisterGlobalConstant) TableName() string {
+	return "register_global_constant"
+}

cmd/mempool/tzkt/tzkt.go

@@ -114,29 +114,16 @@ func (tzkt *TzKT) Blocks() <-chan BlockMessage {
 func (tzkt *TzKT) handleBlockMessage(msg events.Message) error {
 	switch msg.Type {
 	case events.MessageTypeData:
-		data, ok := msg.Body.([]interface{})
-		if !ok {
-			return errors.Wrapf(ErrInvalidBlockType, "%v", msg.Body)
-		}
-		if len(data) == 0 {
+		if msg.Body == nil {
 			return nil
 		}
-		blockData, ok := data[0].(map[string]interface{})
-		if !ok {
-			return errors.Wrapf(ErrInvalidBlockType, "%v", data[0])
-		}
-		hash, err := getString(blockData, "hash")
-		if err != nil {
-			return err
-		}
-		level, err := getUint64(blockData, "level")
-		if err != nil {
-			return err
-		}
-		tzkt.blocks <- BlockMessage{
-			Hash:  hash,
-			Level: level,
-			Type:  msg.Type,
+		blocks := msg.Body.([]events.Block)
+		for i := range blocks {
+			tzkt.blocks <- BlockMessage{
+				Hash:  blocks[i].Hash,
+				Level: blocks[i].Level,
+				Type:  msg.Type,
+			}
 		}
 	case events.MessageTypeState, events.MessageTypeReorg:
 		tzkt.blocks <- BlockMessage{
@@ -153,7 +140,11 @@ func (tzkt *TzKT) handleBlockMessage(msg events.Message) error {
 func (tzkt *TzKT) handleOperationMessage(msg events.Message) error {
 	switch msg.Type {
 	case events.MessageTypeData:
-		return tzkt.handleUpdateMessage(msg)
+		if msg.Body == nil {
+			return nil
+		}
+		operations := msg.Body.([]interface{})
+		return tzkt.handleUpdateMessage(operations)
 	case events.MessageTypeState, events.MessageTypeReorg:
 	default:
 		return errors.Wrapf(ErrUnknownMessageType, "%d", msg.Type)
@@ -162,19 +153,11 @@ func (tzkt *TzKT) handleOperationMessage(msg events.Message) error {
 	return nil
 }
 
-func (tzkt *TzKT) handleUpdateMessage(msg events.Message) error {
+func (tzkt *TzKT) handleUpdateMessage(operations []interface{}) error {
 	message := newOperationMessage()
 
-	body, ok := msg.Body.([]interface{})
-	if !ok {
-		return errors.Wrapf(ErrInvalidBodyType, "%T", msg.Body)
-	}
-	for i := range body {
-		operation, ok := body[i].(map[string]interface{})
-		if !ok {
-			return errors.Wrapf(ErrInvalidOperationType, "%T", body[i])
-		}
-		if err := tzkt.processOperation(operation, &message); err != nil {
+	for i := range operations {
+		if err := tzkt.processOperation(operations[i], &message); err != nil {
 			return err
 		}
 	}
@@ -184,33 +167,81 @@ func (tzkt *TzKT) handleUpdateMessage(msg events.Message) error {
 	return nil
 }
 
-func getString(data map[string]interface{}, key string) (string, error) {
-	value, ok := data[key]
-	if !ok {
-		return "", errors.Wrapf(ErrOperationDoesNotContain, "field=%s data=%v", key, data)
-	}
-	s, ok := value.(string)
-	if !ok {
-		return "", errors.Wrapf(ErrInvalidFieldType, "field=%s expected_type=string type=%T data=%v", key, value, value)
-	}
-	return s, nil
-}
+func (tzkt *TzKT) getAPIOperation(data interface{}) (api.Operation, error) {
+	switch operation := data.(type) {
+
+	case *events.Delegation:
+		tx := api.Operation{
+			ID:       operation.ID,
+			Level:    operation.Level,
+			Hash:     operation.Hash,
+			Kind:     toNodeKinds[operation.Type],
+			Block:    operation.Block,
+			GasUsed:  &operation.GasUsed,
+			BakerFee: &operation.BakerFee,
+		}
+		if operation.NewDelegate != nil {
+			tx.Delegate = &api.Address{
+				Alias:   operation.NewDelegate.Alias,
+				Address: operation.NewDelegate.Address,
+			}
+		}
+		return tx, nil
+
+	case *events.Origination:
+		tx := api.Operation{
+			ID:       operation.ID,
+			Level:    operation.Level,
+			Hash:     operation.Hash,
+			Kind:     toNodeKinds[operation.Type],
+			Block:    operation.Block,
+			GasUsed:  &operation.GasUsed,
+			BakerFee: &operation.BakerFee,
+		}
+		return tx, nil
+
+	case *events.Reveal:
+		tx := api.Operation{
+			ID:       operation.ID,
+			Level:    operation.Level,
+			Hash:     operation.Hash,
+			Kind:     toNodeKinds[operation.Type],
+			Block:    operation.Block,
+			GasUsed:  &operation.GasUsed,
+			BakerFee: &operation.BakerFee,
+		}
+		return tx, nil
+
+	case *events.Transaction:
+		tx := api.Operation{
+			ID:       operation.ID,
+			Level:    operation.Level,
+			Hash:     operation.Hash,
+			Kind:     toNodeKinds[operation.Type],
+			Block:    operation.Block,
+			GasUsed:  &operation.GasUsed,
+			BakerFee: &operation.BakerFee,
+		}
+		if operation.Parameter != nil {
+			tx.Parameters = &api.Parameters{
+				Entrypoint: operation.Parameter.Entrypoint,
+				Value:      operation.Parameter.Value,
+			}
+		}
+		return tx, nil
 
-func getUint64(data map[string]interface{}, key string) (uint64, error) {
-	value, ok := data[key]
-	if !ok {
-		return 0, errors.Wrapf(ErrOperationDoesNotContain, "field=%s data=%v", key, data)
-	}
-	f, ok := value.(float64)
-	if !ok {
-		return 0, errors.Wrapf(ErrInvalidFieldType, "field=%s expected_type=float64 type=%T data=%v", key, value, value)
+	case map[string]interface{}:
+		var general api.Operation
+		err := mapstructure.Decode(data, &general)
+		return general, err
+	default:
+		return api.Operation{}, errors.Wrapf(ErrInvalidOperationType, "%T", data)
 	}
-	return uint64(f), nil
 }
 
-func (tzkt *TzKT) processOperation(data map[string]interface{}, message *OperationMessage) error {
-	var operation api.Operation
-	if err := mapstructure.Decode(data, &operation); err != nil {
+func (tzkt *TzKT) processOperation(data interface{}, message *OperationMessage) error {
+	operation, err := tzkt.getAPIOperation(data)
+	if err != nil {
 		return err
 	}
 	operation.Kind = toNodeKinds[operation.Kind]

go.mod

@@ -3,10 +3,8 @@ module github.com/dipdup-net/mempool
 go 1.15
 
 require (
-	github.com/allegro/bigcache v1.2.1 // indirect
 	github.com/btcsuite/btcutil v1.0.2
-	github.com/dipdup-net/go-lib v0.1.26
-	github.com/gofrs/uuid v4.0.0+incompatible // indirect
+	github.com/dipdup-net/go-lib v0.1.30
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/karlseguin/ccache v2.0.3+incompatible
 	github.com/karlseguin/expect v1.0.8 // indirect
@@ -15,13 +13,12 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.7.0
-	github.com/syndtr/goleveldb v1.0.0 // indirect
 	github.com/ubiq/go-ubiq v3.0.1+incompatible
 	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
-	golang.org/x/sys v0.0.0-20210923061019-b8560ed6a9b7 // indirect
+	golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 	gorm.io/datatypes v1.0.2
-	gorm.io/driver/postgres v1.1.1 // indirect
-	gorm.io/gorm v1.21.15
+	gorm.io/driver/postgres v1.1.2 // indirect
+	gorm.io/gorm v1.21.16
 )