package app import ( "bufio" "bytes" "encoding/json" "io/ioutil" "net/http" "os" "time" "github.com/grvlle/constellation_wallet/backend/models" ) // Transaction 包含所有 tx 信息 type Transaction struct { Edge struct { ObservationEdge struct { Parents []struct { HashReference string `json:"hashReference"` HashType string `json:"hashType"` BaseHash string `json:"baseHash"` } `json:"parents"` Data struct { HashReference string `json:"hashReference"` HashType string `json:"hashType"` BaseHash string `json:"baseHash"` } `json:"data"` } `json:"observationEdge"` SignedObservationEdge struct { SignatureBatch struct { Hash string `json:"hash"` Signatures []struct { Signature string `json:"signature"` ID struct { Hex string `json:"hex"` } `json:"id"` } `json:"signatures"` } `json:"signatureBatch"` } `json:"signedObservationEdge"` Data struct { Amount int64 `json:"amount"` LastTxRef struct { PrevHash string `json:"prevHash"` Ordinal int `json:"ordinal"` } `json:"lastTxRef"` Fee int64 `json:"fee,omitempty"` Salt int64 `json:"salt"` } `json:"data"` } `json:"edge"` LastTxRef struct { PrevHash string `json:"prevHash"` Ordinal int `json:"ordinal"` } `json:"lastTxRef"` IsDummy bool `json:"isDummy"` IsTest bool `json:"isTest"` } /* 发送交易 */ // TriggerTXFromFE 将发起一个从前端触发的新交易。 func (a *WalletApplication) TriggerTXFromFE(amount float64, fee float64, address string) bool { amountConverted := int64(amount * 1e8) feeConverted := int64(fee * 1e8) a.PrepareTransaction(amountConverted, feeConverted, address) for !a.TransactionFinished { time.Sleep(1 * time.Second) } return a.TransactionFailed } // PrepareTransaction 从前端 (Transaction.vue) 触发,并将初始化一个新的 tx。 // 调用的方法在 buildchain.go 中定义 func (a *WalletApplication) PrepareTransaction(amount int64, fee int64, address string) { balance, err := a.GetTokenBalance() if err != nil { a.log.Errorln("查询钱包余额时出错。原因: ", err) a.sendWarning("无法轮询钱包的余额。请稍后重试。") a.TransactionFailed = true return } if amount+fee > int64(balance*1e8) { a.log.Warnf("Trying to send: %d", amount+fee) a.log.Warnf("Insufficient Balance: %d", int64(balance*1e8)) a.sendWarning("Insufficent Balance.") a.TransactionFailed = true return } if a.TransactionFinished { a.TransactionFinished = false // 异步通知 FE 钱包中的 TX 状态。 go func() { for !a.TransactionFinished { a.RT.Events.Emit("tx_in_transit", a.TransactionFinished) time.Sleep(1 * time.Second) } a.RT.Events.Emit("tx_in_transit", a.TransactionFinished) }() ptx := a.loadTXFromFile(a.paths.PrevTXFile) ltx := a.loadTXFromFile(a.paths.LastTXFile) ptxObj, ltxObj := a.convertToTXObject(ptx, ltx) a.formTXChain(amount, fee, address, ptxObj, ltxObj) } } func (a *WalletApplication) putTXOnNetwork(tx *Transaction) (bool, string) { a.log.Info("尝试与 上的主网通信: " + a.Network.URL + a.Network.Handles.Transaction) /* 暂时注释掉 */ a.log.Warnln("TX 序数:", tx.Edge.Data.LastTxRef.Ordinal) bytesRepresentation, err := json.Marshal(tx) if err != nil { a.log.Errorln("无法解析事务的 JSON 数据", err) a.sendError("无法解析事务的 JSON 数据", err) return false, "" } resp, err := http.Post(a.Network.URL+a.Network.Handles.Transaction, "application/json", bytes.NewBuffer(bytesRepresentation)) if err != nil { a.log.Errorln("发送 HTTP 请求失败。原因: ", err) a.sendError("无法向主网发送请求。请检查您的互联网连接。原因: ", err) return false, "" } defer resp.Body.Close() if resp.StatusCode == http.StatusOK { bodyBytes, err := ioutil.ReadAll(resp.Body) if err != nil { a.log.Errorln(string(bodyBytes)) a.log.Errorln("无法读取响应正文。原因: ", err) } bodyString := string(bodyBytes[1:65]) a.log.Infoln("请求正文的 bytesize: ", len(bodyBytes)) if len(bodyBytes) == 66 { a.log.Info("交易哈希: ", bodyString) a.TxPending(bodyString) a.log.Infoln("交易已成功发送到网络。") a.sendSuccess("交易发送成功!") return true, bodyString } a.log.Warn(bodyString) a.sendWarning("无法将交易放到网络上。原因: " + bodyString) return false, "" } bodyBytes, err := ioutil.ReadAll(resp.Body) if err != nil { a.log.Errorln(err) } bodyString := string(bodyBytes) a.sendError("Unable to communicate with mainnet. Reason: "+bodyString, err) a.log.Errorln("Unable to put TX on the network. HTTP Code: " + string(resp.StatusCode) + " - " + bodyString) return false, "" } func (a *WalletApplication) sendTransaction(txFile string) *models.TXHistory { txObject := a.loadTXFromFile(txFile) tx := &Transaction{} bytes := []byte(txObject) err := json.Unmarshal(bytes, &tx) if err != nil { a.sendError("无法解析最后一个事务。原因:", err) a.log.Errorf("无法解析 last_tx 的内容。原因: %s", err) return nil } // Put TX object on network TXSuccessfullyPutOnNetwork, hash := a.putTXOnNetwork(tx) if TXSuccessfullyPutOnNetwork { txData := &models.TXHistory{ Amount: tx.Edge.Data.Amount, Receiver: tx.Edge.ObservationEdge.Parents[1].HashReference, Fee: tx.Edge.Data.Fee, Hash: hash, TS: time.Now().Format("Jan _2 15:04:05"), Status: "Pending", Failed: false, } a.storeTX(txData) a.RT.Events.Emit("new_transaction", txData) // Pass the tx to the frontend as a new transaction. a.TransactionFinished = true a.TransactionFailed = false return txData } txData := &models.TXHistory{ Amount: tx.Edge.Data.Amount, Receiver: tx.Edge.ObservationEdge.Parents[1].HashReference, Fee: tx.Edge.Data.Fee, Hash: hash, TS: time.Now().Format("Jan _2 15:04:05"), Status: "Error", Failed: true, } a.log.Errorln("TX 失败,存储失败状态。") a.storeTX(txData) a.TransactionFinished = true a.TransactionFailed = true return txData } func (a *WalletApplication) storeTX(txData *models.TXHistory) { if txData == nil { return } if err := a.DB.Model(&a.wallet).Where("wallet_alias = ?", a.wallet.WalletAlias).Association("TXHistory").Append(txData).Error; err != nil { a.log.Errorln("无法使用新的 TX 更新 DB 记录。原因: ", err) a.sendError("无法使用新的 TX 更新 DB 记录。原因: ", err) } a.log.Infoln("已成功将 tx 存储在 DB 中") } // loadTXFromFile 获取文件,扫描它并在对象中返回它 func (a *WalletApplication) loadTXFromFile(txFile string) string { var txObjects string fi, err := os.Stat(txFile) if err != nil { a.log.Errorln("无法统计last_tx。原因: ", err) a.sendError("无法统计last_tx。原因: ", err) return "" } // get the size size := fi.Size() if size <= 0 { a.log.Info("TX file is empty.") return "" } file, err := os.Open(txFile) // acct if err != nil { a.log.Errorln("无法打开 TX 文件。原因: ", err) a.sendError("无法读取最后的 tx。中止...原因: ", err) return "" } scanner := bufio.NewScanner(file) scanner.Split(bufio.ScanLines) for scanner.Scan() { txObjects = scanner.Text() } defer file.Close() return txObjects } /* 查询 TX */ // TxProcessed 将查询最后一笔交易。如果未返回任何答案,则表示该答案已处理,并且 method 将返回 true。 func (a *WalletApplication) TxProcessed(TXHash string) bool { a.log.Info("Communicating with mainnet on: " + a.Network.URL + a.Network.Handles.Transaction + "/" + TXHash) resp, err := http.Get(a.Network.URL + a.Network.Handles.Transaction + "/" + TXHash) if err != nil { a.log.Errorln("发送 HTTP 请求失败。原因: ", err) if err := a.DB.Model(&a.wallet).Where("wallet_alias = ?", a.wallet.WalletAlias).Delete(&a.wallet).Error; err != nil { a.log.Errorln("导入失败时无法删除 wallet。原因: ", err) return false } a.log.Errorln("无法验证交易状态。请检查您的互联网连接。") return false } defer resp.Body.Close() if resp.Body == nil { return false } bodyBytes, err := ioutil.ReadAll(resp.Body) if err != nil { return false } // 声明为空接口 var result map[string]interface{} // 将 JSON 解组或解码到接口。 err = json.Unmarshal(bodyBytes, &result) if err != nil { return false } if result["cbBaseHash"] != nil { a.log.Infoln("CheckPoint 哈希:", result["cbBaseHash"]) return true } // null response means it's snapshotted return string(bodyBytes) == "null" } type txStatus struct { Complete string Pending string Error string } // TxPending 获取 TX 哈希并使用当前状态(Pending/Error/Complete)更新前端 func (a *WalletApplication) TxPending(TXHash string) { status := &txStatus{ Complete: "Complete", Pending: "Pending", Error: "Error", } consensus := 0 select { case <-a.killSignal: return default: go func() bool { for retryCounter := 0; retryCounter < 30; retryCounter++ { processed := a.TxProcessed(TXHash) if !processed { a.log.Warnf("Transaction %v pending", TXHash) a.RT.Events.Emit("tx_pending", status.Pending) time.Sleep(time.Duration(retryCounter) * time.Second) // Increase polling interval if retryCounter == 29 { // Register failed transaction a.sendWarning("Unable to get verification of processed transaction from the network. 请稍后重试。") a.log.Errorf("Unable to get status from the network on transaction: %s", TXHash) a.RT.Events.Emit("tx_pending", status.Error) if err := a.DB.Table("tx_histories").Where("hash = ?", TXHash).Updates(map[string]interface{}{"status": status.Error, "failed": true}).Error; err != nil { a.log.Errorln("Unable to query database object for the imported wallet. Reason: ", err) a.LoginError("Unable to query database object for the imported wallet.") return false } a.RT.Events.Emit("update_tx_history", []models.TXHistory{}) // Clear TX history a.initTXFromDB() return false } consensus = 0 // Reset consensus } if processed && consensus != 3 { consensus++ a.log.Infof("TX status check has reached consensus %v/3", consensus) time.Sleep(1 * time.Second) } if processed && consensus == 3 { // Need five consecetive confirmations that TX has been processed. break } } a.log.Infof("已成功处理 %v 交易", TXHash) a.sendSuccess("交易 " + TXHash[:30] + " 已成功处理") if err := a.DB.Table("tx_histories").Where("hash = ?", TXHash).UpdateColumn("status", status.Complete).Error; err != nil { a.log.Errorln("Unable to query database object for the imported wallet. Reason: ", err) a.LoginError("Unable to query database object for the imported wallet.") return false } a.RT.Events.Emit("tx_pending", status.Complete) a.RT.Events.Emit("update_tx_history", []models.TXHistory{}) // Clear TX history a.initTXFromDB() return true }() } }