update code

chain/sync.go

@@ -321,7 +321,10 @@ func (s *ChainSync) FilterValidatorEvents(txLog types.Log, tx *dao.Transaction)
 		inId      = int64(0)
 		eventHash = ""
 		txHash    = txLog.TxHash.Hex()
+		valOp     = constant.TransferChainNoProcess
 	)
+	isMyselfOp := false
+	valAddr := strings.ToLower(s.d.GetChainValidatorAddr(s.chain).Hex())
 
 	abi, _ := bridge.BridgeContractMetaData.GetAbi()
 	switch txLog.Topics[0].Hex() {
@@ -335,6 +338,10 @@ func (s *ChainSync) FilterValidatorEvents(txLog types.Log, tx *dao.Transaction)
 		validator = strings.ToLower(event.Validator.String())
 		inId = event.InId.Int64()
 		eventType = "TransferInConfirmation"
+		valOp = constant.ValidatorStatusConfirmation
+		if validator == valAddr {
+			isMyselfOp = true
+		}
 	case abi.Events["TransferInRejection"].ID.Hex():
 		event, err := s.bridgeCa.ParseTransferInRejection(txLog)
 		if err != nil {
@@ -345,6 +352,13 @@ func (s *ChainSync) FilterValidatorEvents(txLog types.Log, tx *dao.Transaction)
 		validator = strings.ToLower(event.Validator.String())
 		inId = event.InId.Int64()
 		eventType = "TransferInRejection"
+		valOp = constant.ValidatorStatusRejection
+		if validator == valAddr {
+			isMyselfOp = true
+		}
+	default:
+		log.WithField("chain", s.name).Error("unknown event")
+		return nil
 	}
 	err := s.d.CreateValidatorEventTx(tx, eventHash, chainId, validator, txHash, eventType, inId)
 	if err != nil {
@@ -353,6 +367,14 @@ func (s *ChainSync) FilterValidatorEvents(txLog types.Log, tx *dao.Transaction)
 		}).Error("db create validator event")
 		return err
 	}
+	if isMyselfOp {
+		event, _ := s.d.GetBridgeEventWithInInfoTx(tx, chainId, inId)
+		if event != nil {
+			if err = s.d.UpdateBridgeValidatorOperationTx(tx, event, valOp); err != nil {
+				log.WithError(err).Error("db update validator operation event")
+			}
+		}
+	}
 	return nil
 }
 

dao/db.go

@@ -28,7 +28,7 @@ func (d *Dao) SetStorageHeight(key string, intValue int64) (err error) {
 
 func (d *Dao) GetUnprocessedBridgeEvents(limit int) (events []*dbModel.BridgeEvent, err error) {
 	err = d.db.Model(&dbModel.BridgeEvent{}).
-		Where("`to_chain_status` = ? AND `validator_status` = ?", 0, 0).
+		Where("`to_chain_status` < ? AND `validator_status` = ?", 2, 0).
 		Order("created_at ASC").
 		Limit(limit).
 		Find(&events).Error

dao/dbtx.go

@@ -68,6 +68,12 @@ func (d *Dao) UpdateBridgeWithTransferInTx(tx *Transaction, event *dbModel.Bridg
 	}).Error
 }
 
+func (d *Dao) UpdateBridgeValidatorOperationTx(tx *Transaction, event *dbModel.BridgeEvent, op int) (err error) {
+	return tx.tx.Model(&dbModel.BridgeEvent{}).Where("`id` = ?", event.ID).Updates(map[string]interface{}{
+		"validator_status": op,
+	}).Error
+}
+
 func (d *Dao) UpdateBridgeResultTx(tx *Transaction, event *dbModel.BridgeEvent, toChainHash string, status int) error {
 	return tx.tx.Model(&dbModel.BridgeEvent{}).Where("`id` = ?", event.ID).Updates(map[string]interface{}{
 		"to_chain_status": status,

dao/eth.go

@@ -247,3 +247,12 @@ func (d *Dao) HandleTasks() {
 		}
 	}
 }
+
+func (d *Dao) GetChainValidatorAddr(chain *config.ChainConfig) common.Address {
+	k := chain.ValidatorPrivateKey
+	private, err := crypto.ToECDSA(common.FromHex(k))
+	if err != nil {
+		return common.Address{}
+	}
+	return crypto.PubkeyToAddress(private.PublicKey)
+}