validator.go

package validator

import (
	"context"
	"encoding/json"
	"example.com/m/conf"
	"example.com/m/db"
	"example.com/m/log"
	"example.com/m/operate"
	nodemanagerv1 "github.com/odysseus/odysseus-protocol/gen/proto/go/nodemanager/v1"
	witnessV1 "github.com/odysseus/odysseus-protocol/gen/proto/go/witness/v1"
	"google.golang.org/grpc"
	"math/rand"
	"time"
)

type ProofWorker struct {
	productProofChan chan *witnessV1.Proof
	consumeProofChan chan []*witnessV1.Proof
	isCommitProof    map[string]bool
}

func NewProofWorker() *ProofWorker {
	return &ProofWorker{
		productProofChan: make(chan *witnessV1.Proof, 1000),
		consumeProofChan: make(chan []*witnessV1.Proof, 1000),
		isCommitProof:    make(map[string]bool, 0),
	}
}

func (p *ProofWorker) ProductProof(nmResultMsg *nodemanagerv1.ProofTaskResult, taskType uint64, reqHash []byte, respHash []byte, containerSign, minerSign []byte) {
	log.Info("ProductProof received workLoad:", nmResultMsg.Workload)
	p.productProofChan <- &witnessV1.Proof{
		Workload:           nmResultMsg.Workload,
		TaskId:             nmResultMsg.TaskId,
		ReqHash:            reqHash,
		RespHash:           respHash,
		ManagerSignature:   nmResultMsg.ManagerSignature,
		ContainerSignature: containerSign,
		MinerSignature:     minerSign,
		TaskType:           taskType,
		Timestamp:          nmResultMsg.Timestamp,
	}
}

func (p *ProofWorker) ProofStorage() {
	go func(productProofChan chan *witnessV1.Proof) {
		for {
			select {
			case proof := <-productProofChan:
				{
					proofByte, err := json.Marshal(proof)
					if err != nil {
						log.Error("Failed to marshal proof: ", err)
						continue
					}
					err = db.Put(proof.TaskId, proofByte)
					if err != nil {
						log.Error("leveldb put proof failed: ", err)
						continue
					}
					p.isCommitProof[proof.TaskId] = false
				}
			}
		}
	}(p.productProofChan)

	timer := time.NewTicker(time.Minute)
	randomMinute := getRandInt()
	for {
		select {
		case <-timer.C:
			nowTime := time.Now()
			min := nowTime.Minute()
			if min == 0 {
				randomMinute = getRandInt()
			}
			if nowTime.Hour() == 23 {
				randomMinute = 59
			}
			// 检查是否在指定时间范围内（40-59分钟）
			if min >= 40 && min <= 59 && min == randomMinute {
				proofs := make([]*witnessV1.Proof, 0)
				iter, err := db.NewIterator()
				if err != nil {
					log.Error("db new iterator failed: ", err)
					continue
				}
				if iter == nil {
					log.Warn("level db iterator is nil")
					continue
				}
				for iter.Next() {
					proof := &witnessV1.Proof{}
					err := json.Unmarshal(iter.Value(), proof)
					if err != nil {
						log.Error("Error parsing proof from database: ", err)
						continue
					}
					if p.isCommitProof[proof.TaskId] {
						continue
					}
					p.isCommitProof[proof.TaskId] = true
					proofs = append(proofs, proof)
					//err = db.Delete(iter.Key())
					//if err != nil {
					//	log.Error("Error deleting proof from database: ", err)
					//	return
					//}
				}
				if len(proofs) > 0 {
					p.consumeProofChan <- proofs
					log.Info("---------------------------Storage proof data---------------------------")
				}
			}
		}
	}
}

func (p *ProofWorker) CommitWitness() {
	validatorClient := operate.ConnValidatorGrpc(conf.GetConfig().ValidatorUrl)
	for {
		select {
		case proofs := <-p.consumeProofChan:
			proofsReq := &witnessV1.PushProofRequest{
				Proofs:        proofs,
				MinerAddress:  conf.GetConfig().SignPublicAddress.Hex(),
				RewardAddress: conf.GetConfig().BenefitAddress,
			}
			pushProof, err := validatorClient.PushProof(context.Background(), proofsReq, grpc.EmptyCallOption{})
			if err != nil {
				log.Error("Push proof failed :", err)
				continue
			}
			workload := pushProof.GetWorkload()
			log.Info("Commit proof time:", time.Now())
			log.Info("Push proof response received : %v", workload)
			log.Info("---------------------------Commit witness data---------------------------")
		}
	}
}

func getRandInt() int {
	return rand.Intn(20) + 40
}