[SOT-197] feat: add verifier (#2)

* feat: add verifier

* fix: update log

* doc: update README.md

* fix: create one ipfs shell.NewShell() session per QmHash

* fix: add output-timestamp dir for verification output

Author: zyyw

README.md

@@ -1,2 +1,43 @@
 # btfs-migration-tools
-tools: migrate ipfs file to btfs network
+> This is a command line tool to migrate ipfs file to btfs network
+
+## Get the source code
+```bash
+go get -u github.com/TRON-US/btfs-migration-toolkit
+```
+
+## Build the binary
+```bash
+go build
+```
+
+## Usage
+The file migration from IPFS network to BTFS network has to be done in two phases. 
+
+Phase one: download a file from IPFS to local file system and upload this file to BTFS network 
+through Soter using soter-go-sdk. 
+
+Phase two: verify a file has indeed been stored in BTFS network by BTT Integration Architecture.
+This is achieved by using the request id to query soter status record.
+
+Before using this command line tool, users should configure the config.yaml file first and provide a input.csv file.
+### Example
+#### Batch upload
+```bash
+./btfs-migration-toolkit --config=config.yaml --method=batch_upload --input=input.csv
+```
+
+#### Single upload
+```bash
+./btfs-migration-toolkit --config=config.yaml --method=single_upload --hash=Qmxxxxxx
+```
+
+#### Batch verify
+```bash
+./btfs-migration-toolkit --config=config.yaml --method=batch_verify --input=output_hash.csv
+```
+
+#### Single verify
+```bash
+./btfs-migration-toolkit --config=config.yaml --method=single_verify --request-id=xxxxx
+```

constants/constant.go

@@ -4,10 +4,29 @@ const (
 	OkCode                  = 0
 	InsufficientBalanceCode = 20009
 
-	OutputHashFileName  = "output_hash.csv"
-	OutputRetryFileName = "output_retry.csv"
+	// files has been downloaded from IPFS network and uploaded to BTFS network through Soter
+	// BTFS hash and request id have been return from Soter and saved to this file
+	OutputHashFileName    = "output_hash.csv"
+	// the migration has been failed, and IPFS hash is saved in this file for retry
+	OutputRetryFileName   = "output_retry.csv"
+	// record identified by this request id is in Pending status
+	OutputPendingFileName = "output_P.csv"
+	// record identified by this request id is in Failed status
+	OutputFailFileName    = "output_F.csv"
+	// record identified by this request id is in Success status
+	OutputSucessFileName  = "output_S.csv"
+	// this request id is invalid or verification is abort due to internal error
+	OutputErrorFileName   = "output_E.csv"
 
 	// method
-	BatchUpload = "batch_upload"
+	BatchUpload  = "batch_upload"
 	SingleUpload = "single_upload"
+	BatchVerify  = "batch_verify"
+	SingleVerify = "single_verify"
+
+	//
+	Delimiter = ","
+	StatusP   = "P"
+	StatusF   = "F"
+	StatusS   = "S"
 )

core/load_config.go

@@ -2,14 +2,11 @@ package core
 
 import (
 	"github.com/TRON-US/btfs-migration-toolkit/conf"
-
 	"github.com/fsnotify/fsnotify"
-	"github.com/ipfs/go-ipfs-api"
 	"github.com/spf13/viper"
 )
 
 var Conf *conf.Config
-var Sh *shell.Shell
 
 //init config file to viper
 func InitConfig(cfg string) error {
@@ -48,7 +45,6 @@ func getNewConfig() error {
 	if err := viper.Unmarshal(Conf); err != nil {
 		return err
 	}
-	Sh = shell.NewShell(Conf.IpfsUrl)
 
 	return nil
 }

core/soter_response.go

@@ -1,6 +1,16 @@
 package core
 
-type SoterResponse struct {
+type SoterAddFileResponse struct {
 	Cid       string `json:"cid"`
 	RequestId string `json:"request_id"`
 }
+
+type SoterOrderDetailsResponse struct {
+	FileHash    string `json:"file_hash"`
+	FileSize    int64  `json:"file_size"`
+	FileName    string `json:"file_name"`
+	Fee         int64  `json:"fee"`
+	Type        string `json:"type"`
+	Description string `json:"description"`
+	Status      string `json:"status"`
+}

main.go

@@ -2,10 +2,11 @@ package main
 
 import (
 	"fmt"
+	"os"
+
 	"github.com/TRON-US/btfs-migration-toolkit/constants"
 	"github.com/TRON-US/btfs-migration-toolkit/core"
-	"github.com/TRON-US/btfs-migration-toolkit/uploader"
-	"os"
+	"github.com/TRON-US/btfs-migration-toolkit/service"
 
 	"github.com/spf13/pflag"
 )
@@ -15,6 +16,7 @@ var (
 	method = pflag.StringP("method", "m", "", "choose a method to run")
 	inputFile = pflag.StringP("input", "i", "", "input .csv file with a list of IPFS QmHash")
 	ipfsHash = pflag.StringP("hash", "h", "", "IPFS hash to migrate")
+	requestId = pflag.StringP("request-id", "r", "", "request id to query for")
 )
 
 func main() {
@@ -26,16 +28,15 @@ func main() {
 
 	switch *method {
 	case constants.BatchUpload:
-		uploader.BatchUpload(*inputFile)
+		service.BatchUpload(*inputFile)
 	case constants.SingleUpload:
-		uploader.SingleUpload(*ipfsHash)
+		service.SingleUpload(*ipfsHash)
+	case constants.BatchVerify:
+		service.BatchVerify(*inputFile)
+	case constants.SingleVerify:
+		service.SingleVerify(*requestId)
 	default:
 		fmt.Println("unknown method")
 		os.Exit(0)
 	}
-
-	fmt.Printf("Migration complete.\n" +
-		"Please checkout %s and %s for batch migration.\n" +
-		"Or review screen output for single migration.\n",
-		constants.OutputHashFileName, constants.OutputRetryFileName)
 }

uploader/upload.go renamed to service/uploader.go

@@ -1,21 +1,22 @@
-package uploader
+package service
 
 import (
 	"bufio"
 	"encoding/json"
 	"fmt"
+	"os"
+	"strings"
+	"sync"
+
 	"github.com/TRON-US/btfs-migration-toolkit/constants"
 	"github.com/TRON-US/btfs-migration-toolkit/core"
 	"github.com/TRON-US/btfs-migration-toolkit/log"
 	"github.com/TRON-US/soter-sdk-go/soter"
-	"os"
-	"strings"
-	"sync"
+
+	"github.com/ipfs/go-ipfs-api"
 )
 
 func BatchUpload(inputFilename string) {
-	batchSize := core.Conf.BatchSize
-
 	inputHashFile, err := os.Open(inputFilename)
 	if err != nil {
 		log.Logger().Error(fmt.Sprintf("Failed to open file %s, reason=[%v]", inputFilename, err))
@@ -27,7 +28,7 @@ func BatchUpload(inputFilename string) {
 		}
 	}()
 
-	outputHashFile, err := os.Create(fmt.Sprintf("./%s", constants.OutputHashFileName))
+	outputHashFile, err := os.Create(constants.OutputHashFileName)
 	if err != nil {
 		log.Logger().Error(fmt.Sprintf("Failed to open file %s, reason=[%v]", constants.OutputHashFileName, err))
 		os.Exit(1)
@@ -38,10 +39,18 @@ func BatchUpload(inputFilename string) {
 				constants.OutputHashFileName, err))
 		}
 	}()
-	outputRetryFile, err := os.Create(fmt.Sprintf("./%s", constants.OutputRetryFileName))
+
+	outputRetryFile, err := os.Create(constants.OutputRetryFileName)
 	if err != nil {
-		log.Logger().Error(fmt.Sprintf("Failed to open %s, reason=[%v]", constants.OutputRetryFileName, err))
+		log.Logger().Error(fmt.Sprintf("Failed to open file %s, reason=[%v]", constants.OutputRetryFileName, err))
+		os.Exit(1)
 	}
+	defer func() {
+		if err := outputRetryFile.Close(); err != nil {
+			log.Logger().Error(fmt.Sprintf("Failed to close file %s, reason=[%v]",
+				constants.OutputRetryFileName, err))
+		}
+	}()
 
 	wg := sync.WaitGroup{}
 	scanner := bufio.NewScanner(inputHashFile)
@@ -50,44 +59,50 @@ func BatchUpload(inputFilename string) {
 		hash := scanner.Text()
 		wg.Add(1)
 		counter++
-		go func(h string, outFile *os.File, retryFile *os.File) {
+		go func(h string, outFile, retryFile *os.File) {
 			defer wg.Done()
 			res, err := migrate(h)
 			if err != nil {
-				log.Logger().Error(fmt.Sprintf("ipfs_hash=%s, reason=[%v]", h, err))
+				log.Logger().Error(fmt.Sprintf("[ipfs_hash=%s] Failed to migrate, reason=[%v]", h, err))
 				// definitely failed to upload through soter; write to output_retry.csv
 				_, err = fmt.Fprintln(retryFile, h)
 				if err != nil {
-					errMsg := fmt.Sprintf("Failed to write to file %s, hash=%s, reason=[%v]",
-						constants.OutputRetryFileName, h, err)
+					errMsg := fmt.Sprintf("[ipfs_hash=%s] Failed to write to file %s, reason=[%v]",
+						h, constants.OutputRetryFileName, err)
 					log.Logger().Error(errMsg)
 				}
 				return
 			}
-			log.Logger().Debug(fmt.Sprintf("[%s,%s,%s]", h, res[0], res[1]))
-			// write <ipfs_hash, request_id, btfs_hash> to output_hash.csv
+			// write (ipfs_hash, request_id, btfs_hash) to output_hash.csv
+			log.Logger().Debug(fmt.Sprintf("[ipfs_hash=%s] Write to file %s, (%s,%s,%s)",
+				h, constants.OutputHashFileName, h, res[0], res[1]))
 			line := fmt.Sprintf("%s,%s,%s", h, res[0], res[1])
 			_, err = fmt.Fprintln(outFile, line)
 			if err != nil {
 				log.Logger().Error(err.Error())
 			}
 		}(hash, outputHashFile, outputRetryFile)
-		if counter % batchSize == 0 {
+		if counter % core.Conf.BatchSize == 0 {
 			wg.Wait()
 			counter = 0
 		}
 	}
 	// wait here because counter < batchSize and no more lines to read
 	wg.Wait()
 	if err := scanner.Err(); err != nil {
-		log.Logger().Error(err.Error())
+		errMsg := fmt.Sprintf("Failed to scan input file, reason=[%v]", err)
+		log.Logger().Error(errMsg)
 	}
+
+	fmt.Printf("\nMigration complete.\n" +
+		"Please checkout %s and %s for batch migration.\n",
+		constants.OutputHashFileName, constants.OutputRetryFileName)
 }
 
 func SingleUpload(ipfsHash string)  {
 	res, err := migrate(ipfsHash)
 	if err != nil {
-		log.Logger().Error(err.Error())
+		log.Logger().Error(fmt.Sprintf("[ipfs_hash=%s] Failed to migrate, reason=[%v]", ipfsHash, err))
 		os.Exit(1)
 	}
 	fmt.Printf("IPFS hash: %s\n", ipfsHash)
@@ -97,12 +112,11 @@ func SingleUpload(ipfsHash string)  {
 
 func migrate(ipfsHash string) ([]string, error) {
 	if !strings.HasPrefix(ipfsHash, "Qm") {
-		errMsg := fmt.Sprintf("input with invalid IPFS hash [%s]", ipfsHash)
-		log.Logger().Debug(errMsg)
+		errMsg := fmt.Sprintf("[ipfs_hash=%s] Input with invalid ipfs hash", ipfsHash)
 		return nil, fmt.Errorf(errMsg)
 	}
 	// download file from IPFS network to local file system
-	log.Logger().Debug(fmt.Sprintf("downloading file from IPFS network, %s", ipfsHash))
+	log.Logger().Debug(fmt.Sprintf("[ipfs_hash=%s] Downloading the file from IPFS network", ipfsHash))
 	if err := downloadFromIPFS(ipfsHash); err != nil {
 		return nil, err
 	}
@@ -111,12 +125,13 @@ func migrate(ipfsHash string) ([]string, error) {
 	defer func(h string) {
 		// delete local files
 		if err := os.Remove(fmt.Sprintf("./%s", h)); err != nil {
-			errMsg := fmt.Sprintf("Failed to delete file %s", h)
+			errMsg := fmt.Sprintf("[ipfs_hash=%s] Failed to delete local file", h)
 			log.Logger().Error(errMsg)
 		}
 	}(ipfsHash)
 
 	// upload the file to BTFS through soter
+	log.Logger().Debug(fmt.Sprintf("[ipfs_hash=%s] Uploading the file to BTFS network", ipfsHash))
 	res, err := uploadToBTFS(ipfsHash)
 	if err != nil {
 		return nil, err
@@ -126,8 +141,9 @@ func migrate(ipfsHash string) ([]string, error) {
 
 func downloadFromIPFS(hash string) error {
 	// go-ipfs-api, sdk: get
-	if err := core.Sh.Get(hash, hash); err != nil {
-		return err
+	sh := shell.NewShell(core.Conf.IpfsUrl)
+	if err := sh.Get(hash, hash); err != nil {
+		return fmt.Errorf("downloading from IPFS errors out, %v", err)
 	}
 
 	return nil
@@ -138,27 +154,24 @@ func uploadToBTFS(filename string) ([]string, error) {
 	filePath := fmt.Sprintf("./%s", filename)
 	resp, err := sh.AddFile(core.Conf.UserAddress, filePath)
 	if err != nil {
-		log.Logger().Error(err.Error())
-		return nil, err
+		errMsg := fmt.Sprintf("failed to add file, reason=[%v]", err)
+		return nil, fmt.Errorf(errMsg)
 	}
 	if resp.Code != constants.OkCode {
-		errMsg := fmt.Sprintf("Error: code=%d, message=%s", resp.Code, resp.Message)
-		log.Logger().Error(errMsg)
+		errMsg := fmt.Sprintf("response code error: code=%d, message=%s", resp.Code, resp.Message)
 		if resp.Code == constants.InsufficientBalanceCode {
 			os.Exit(1)
 		}
 		return nil, fmt.Errorf(errMsg)
 	}
 	s, err := json.Marshal(resp.Data)
 	if err != nil {
-		log.Logger().Error(err.Error())
-		return nil, err
+		return nil, fmt.Errorf("failed to marshal response data, %v", err)
 	}
-	var soterResponse core.SoterResponse
+	var soterResponse core.SoterAddFileResponse
 	err = json.Unmarshal(s, &soterResponse)
 	if err != nil {
-		log.Logger().Error(err.Error())
-		return nil, err
+		return nil, fmt.Errorf("failed to unmarshal soter response data, %v", err)
 	}
 	res := [...]string{soterResponse.RequestId, soterResponse.Cid}
 	return res[:], nil