Merge pull request #94 from 0xcregis/release-0.5.3

feat: Release 0.5.3

README.md

@@ -77,8 +77,8 @@ Allow all developers to use on-chain services without deploying nodes or knowing
 
 ### 3. Init Config & Database
 
-- ./scripts : Database initialization script, mainly modify the database name (database) default: ether2
-- ./config : The configuration file required for system startup, the following files and fields often need to be modified
+- ./build/scripts : Database initialization script, mainly modify the database name (database) default: ether2
+- ./build/config : The configuration file required for system startup, the following files and fields often need to be modified
   - blockchain_config.json, collect_config.json: mainly modify the node address of each public chain
   - store_config.json : DbName field
   - task_config.json : BlockMin, BlockMax and other fields

README_zh.md

@@ -75,8 +75,8 @@
 
 ### 3. 初始化配置文件和数据库脚本
 
-- ./scripts :数据库初始化脚本,主要修改数据库名称即可（database）默认:ether2
-- ./config :系统启动需要的配置文件,以下文件和字段常需要修改
+- ./build/scripts :数据库初始化脚本,主要修改数据库名称即可（database）默认:ether2
+- ./build/config :系统启动需要的配置文件,以下文件和字段常需要修改
     - blockchain_config.json,collect_config.json :主要修改各个公链的节点地址
     - store_config.json ： DbName 字段
     - task_config.json ：BlockMin、BlockMax等字段

cmd/blockchain/app.go

@@ -15,7 +15,7 @@ import (
 
 func main() {
 	var configPath string
-	flag.StringVar(&configPath, "blockchain", "./cmd/blockchain/config_tron.json", "The system file of config")
+	flag.StringVar(&configPath, "blockchain", "./cmd/blockchain/config_xrp.json", "The system file of config")
 	flag.Parse()
 	if len(configPath) < 1 {
 		panic("can not find config file")

cmd/blockchain/config_filecoin.json

@@ -9,7 +9,7 @@
   },
   "Nodes": [
     {
-      "BlockChain": 301,
+      "BlockChain": 2307,
       "NodeUrl": "https://api.node.glif.io/rpc/v0",
       "NodeToken": ""
     }

cmd/blockchain/config_xrp.json

@@ -9,7 +9,7 @@
   },
   "Nodes": [
     {
-      "BlockChain": 310,
+      "BlockChain": 144,
       "NodeUrl": "https://xrplcluster.com",
       "NodeToken": ""
     }

cmd/collect/README.md

@@ -39,6 +39,12 @@ collect是easynode系统的基础和核心服务，是其他服务的运行的
     "Path": "./log/collect", //日志路径
     "Delay": 2 //日志保存时长（单位：天）
   },
+  "KeyPath": "./data/temp/1", //nodeId的存放路径
+  "Retry": { //重试时间间隔
+    "ErrTx": 100,
+    "NodeTask": 100,
+    "Contract": 100
+  },  
   "Chains": [ //支持公链配置
     {
       "TaskKafka": { //任务kafka
@@ -58,8 +64,8 @@ collect是easynode系统的基础和核心服务，是其他服务的运行的
       },
       "BlockChainName": "eth", //公链昵称
       "BlockChainCode": 200, //公链代码
-      "PullReceipt": false, //是否自动产生收据任务
-      "PullTx": true, //是否自动产生交易任务
+      "PullReceipt": 0, //0:没有收据任务，1:批量收据任务，2:单笔收据任务
+      "PullTx": 1, //0:没有交易任务，1:批量交易任务，2:单笔交易任务
       "BlockTask": { //区块任务配置，非必须，如果未配置，不执行该类的任务
         "FromCluster": [ //区块任务公链节点配置集合
           {

cmd/store/config_tron.json

@@ -2,7 +2,7 @@
   "RootPath": "/api/store",
   "Port": 9003,
   "BaseDb": {
-    "Addr": "192.168.2.9",
+    "Addr": "192.168.2.11",
     "Port": 9000,
     "User": "test",
     "Password": "test",
@@ -21,39 +21,39 @@
       "BackupTxStore": true,
       "Kafka": {
         "BackupTx": {
-          "Host": "192.168.2.9",
+          "Host": "192.168.2.11",
           "Port": 9092,
           "Topic": "backup_tx",
           "Group": "1",
           "StartOffset": -1,
           "Partition": 0
         },
         "SubTx": {
-          "Host": "192.168.2.9",
+          "Host": "192.168.2.11",
           "Port": 9092,
           "Topic": "tron_sub_tx",
           "Group": "110",
           "StartOffset": -1,
           "Partition": 0
         },
         "Tx": {
-          "Host": "192.168.2.9",
+          "Host": "192.168.2.11",
           "Port": 9092,
           "Topic": "tron_tx",
           "Group": "108",
           "StartOffset": -1,
           "Partition": 0
         },
         "Block": {
-          "Host": "192.168.2.9",
+          "Host": "192.168.2.11",
           "Port": 9092,
           "Topic": "tron_block",
           "Group": "103",
           "StartOffset": -1,
           "Partition": 0
         },
         "Receipt": {
-          "Host": "192.168.2.9",
+          "Host": "192.168.2.11",
           "Port": 9092,
           "Topic": "tron_receipt",
           "Group": "2",
@@ -62,7 +62,7 @@
         }
       },
       "ChainDb": {
-        "Addr": "192.168.2.9",
+        "Addr": "192.168.2.11",
         "Port": 9000,
         "User": "test",
         "Password": "test",
@@ -74,7 +74,7 @@
         "BackupTxTable": "backup_tx"
       },
       "Redis": {
-        "Addr": "192.168.2.9",
+        "Addr": "192.168.2.11",
         "Port": 6379,
         "DB": 0
       }

collect/service/cmd/cmd.go

@@ -145,7 +145,7 @@ func (c *Cmd) ReadNodeTaskFromKafka(nodeId string, blockChain int, blockCh chan
 			continue
 		}
 		msg := <-receiver
-		log := c.log.WithFields(logrus.Fields{"model": "Dispatch", "id": time.Now().UnixMilli()})
+		log := c.log.WithFields(logrus.Fields{"model": "Dispatch", "id": time.Now().UnixNano()})
 		task := collect.NodeTask{}
 		err := json.Unmarshal(msg.Value, &task)
 		if err != nil {
@@ -226,7 +226,7 @@ func (c *Cmd) HandlerKafkaRespMessage(msList []*kafka.Message) {
 	//msList := <-kafkaRespCh
 	ids := make([]string, 0, 20)
 	log := c.log.WithFields(logrus.Fields{
-		"id":    time.Now().UnixMilli(),
+		"id":    time.Now().UnixNano(),
 		"model": "HandlerKafkaRespMessage",
 	})
 
@@ -262,7 +262,7 @@ func (c *Cmd) HandlerKafkaRespMessage(msList []*kafka.Message) {
 	}
 
 	if len(ids) > 0 {
-		log.Errorf("UpdateNodeTaskStatus|ids=%v", ids)
+		log.Printf("UpdateNodeTaskStatus|ids=%v", ids)
 		err := c.taskStore.UpdateNodeTaskStatusWithBatch(ids, 1)
 		if err != nil {
 			log.Errorf("UpdateNodeTaskStatusWithBatch|err=%v", err)
@@ -279,7 +279,7 @@ func (c *Cmd) ExecReceiptTask(receiptChan chan *collect.NodeTask, kf chan []*kaf
 		//执行任务
 		receiptTask := <-receiptChan
 		log := c.log.WithFields(logrus.Fields{
-			"id":    time.Now().UnixMilli(),
+			"id":    time.Now().UnixNano(),
 			"model": "ExecReceiptTask",
 		})
 		//log.Printf("receiptTask=%+v", receiptTask)
@@ -389,7 +389,7 @@ func (c *Cmd) ExecTxTask(txCh chan *collect.NodeTask, kf chan []*kafka.Message)
 		//执行任务
 		txTask := <-txCh
 		log := c.log.WithFields(logrus.Fields{
-			"id":    time.Now().UnixMilli(),
+			"id":    time.Now().UnixNano(),
 			"model": "ExecTxTask",
 		})
 
@@ -576,7 +576,7 @@ func (c *Cmd) ExecBlockTask(blockCh chan *collect.NodeTask, kf chan []*kafka.Mes
 		//执行任务
 		blockTask := <-blockCh
 		log := c.log.WithFields(logrus.Fields{
-			"id":    time.Now().UnixMilli(),
+			"id":    time.Now().UnixNano(),
 			"model": "ExecBlockTask",
 		})
 		//log.Printf("taskBlock=%+v", blockTask)

common/chain/chain.go

@@ -19,10 +19,18 @@ import (
 //	"XRP":     {310: 1},
 //}
 
+/**
+
+  arb:63
+  op:64
+  aval:65
+  base:66
+*/
+
 // main
 var defaultChainCode = map[string]map[int64]int8{
 	"ETH":     {60: 1, 6001: 1},
-	"POLYGON": {62: 1, 6201: 1},
+	"POLYGON": {62: 1, 6201: 1, 63: 1, 64: 1, 65: 1, 66: 1},
 	"BSC":     {2510: 1, 2610: 1},
 	"TRON":    {195: 1, 198: 1},
 	"BTC":     {0: 1, 1: 1},

store/db/ch_service.go

@@ -113,6 +113,15 @@ func (m *ClickhouseDb) GetAddressByToken2(token string) ([]*store.MonitorAddress
 	return list, nil
 }
 
+func (m *ClickhouseDb) GetAddressByToken4(token string) ([]*store.MonitorAddress, error) {
+	var list []*store.MonitorAddress
+	err := m.baseDb.Table(m.baseConfig.BaseDb.AddressTable).Select("*").Where("token=?", token).Scan(&list).Error
+	if err != nil || len(list) < 1 {
+		return nil, errors.New("no record")
+	}
+	return list, nil
+}
+
 func (m *ClickhouseDb) DelMonitorAddress(blockchain int64, token string, address string) error {
 	err := m.baseDb.Table(m.baseConfig.BaseDb.AddressTable).Where("token=? and address=? and block_chain=?", token, address, blockchain).Delete(store.MonitorAddress{}).Error
 	if err != nil {
@@ -243,6 +252,10 @@ func (m *ClickhouseDb) AddMonitorAddress(blockchain int64, address *store.Monito
 	return m.baseDb.Table(m.baseConfig.BaseDb.AddressTable).Create(address).Error
 }
 
+func (m *ClickhouseDb) UpdateMonitorAddress(id int64, address *store.MonitorAddress) error {
+	return m.baseDb.Table(m.baseConfig.BaseDb.AddressTable).Where("id=?", id).UpdateColumns(address).Error
+}
+
 func (m *ClickhouseDb) GetAddressByToken(blockchain int64, token string) ([]*store.MonitorAddress, error) {
 	var list []*store.MonitorAddress
 	err := m.baseDb.Table(m.baseConfig.BaseDb.AddressTable).Select("address").Where("token=? and block_chain in (?)", token, []int64{blockchain, 0}).Group("address").Scan(&list).Error

store/db/ch_service_test.go

@@ -2,16 +2,18 @@ package db
 
 import (
 	"log"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/0xcregis/easynode/common/util"
 	"github.com/0xcregis/easynode/store"
 	"github.com/0xcregis/easynode/store/config"
 	"github.com/sunjiangjun/xlog"
 )
 
 func Init() store.DbStoreInterface {
-	cfg := config.LoadConfig("./../../cmd/store/config.json")
+	cfg := config.LoadConfig("./../../cmd/store/config_tron.json")
 	return NewChService(&cfg, xlog.NewXLogger())
 }
 
@@ -51,7 +53,48 @@ func TestClickhouseDb_GetNodeTokenByEmail(t *testing.T) {
 
 func TestClickhouseDb_GetAddressByToken2(t *testing.T) {
 	s := Init()
-	log.Println(s.GetAddressByToken2("36ee0ad5-f4bc-4bca-a1dc-c51db006e249"))
+	//log.Println(s.GetAddressByToken2("36ee0ad5-f4bc-4bca-a1dc-c51db006e249"))
+
+	list, err := s.GetAddressByToken4("90c232ef-ecea-4956-9132-325ebec46e86")
+	if err != nil {
+		t.Error(err)
+	} else {
+		for _, v := range list {
+			if !strings.HasPrefix(v.Address, "0x") {
+				base58Addr, err := util.Base58ToAddress(v.Address)
+				if err != nil {
+					t.Error(err)
+					continue
+				}
+				v.Address = base58Addr.Hex()
+				err = s.UpdateMonitorAddress(v.Id, v)
+				if err != nil {
+					t.Error(err)
+				} else {
+					t.Log("ok")
+				}
+			}
+
+		}
+	}
+
+}
+
+func TestClickhouseDb_UpdateMonitorAddress(t *testing.T) {
+	s := Init()
+	addr := &store.MonitorAddress{
+		Address:    "0x4193465903d04f7890bb9b0b856ab31fc2929ab653",
+		Id:         1702881171946676000,
+		Token:      "73e89458-17ce-48c5-ad5a-58b9bab121c9",
+		TxType:     "0",
+		BlockChain: 198,
+	}
+	err := s.UpdateMonitorAddress(1702881171946676000, addr)
+	if err != nil {
+		t.Error(err)
+	} else {
+		t.Log("ok")
+	}
 }
 
 func TestClickhouseDb_DelMonitorAddress(t *testing.T) {

store/interface.go

@@ -6,9 +6,11 @@ type DbStoreInterface interface {
 	GetNodeTokenByEmail(email string) (*NodeToken, error)
 	GetNodeTokenByToken(token string) (*NodeToken, error)
 	AddMonitorAddress(blockchain int64, address *MonitorAddress) error
+	UpdateMonitorAddress(id int64, address *MonitorAddress) error
 	GetAddressByToken(blockchain int64, token string) ([]*MonitorAddress, error)
 	GetAddressByToken2(token string) ([]*MonitorAddress, error)
 	GetAddressByToken3(blockchain int64) ([]*MonitorAddress, error)
+	GetAddressByToken4(token string) ([]*MonitorAddress, error)
 	DelMonitorAddress(blockchain int64, token string, address string) error
 	NewTx(blockchain int64, tx []*Tx) error
 	NewBlock(blockchain int64, block []*Block) error