package ether

import (
	"math"

	"github.com/coreos/etcd/clientv3"
	"github.com/juju/errors"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	"github.com/pingcap/wormhole/pkg/whpb"
)

const (
	kvRangeLimit = 100000
)

var (
	errTxnFailed = errors.New("[ether] failed to commit transaction")
)

// job 1    -> /clsuetr_path/job/1
// worker 2 -> /clsuetr_path/worker/2
// contract between  worker 1 and job 2 -> /clsuetr_path/contract/1/2
// fence 3    -> /clsuetr_path/fence/3
// pipeline 4 -> /clsuetr_path/pipelines/4

// kv wraps all kv operations, keep it stateless.
type kv struct {
	*etcdutil.Client
	path string
}

func newKV(s *Server) *kv {
	kv := &kv{
		path: s.getClusterRootPath(),
	}
	txn := func(cs ...clientv3.Cmp) clientv3.Txn {
		return s.leaderTxn(cs...)
	}
	kv.Client = etcdutil.NewTxnClient(s.client, kv.path, txn)

	return kv
}

func (kv *kv) loadJobs(jobs *jobsInfo) error {
	nextID := uint64(0)
	endJob := etcdutil.JobPath(kv.path, math.MaxUint64)
	withRange := clientv3.WithRange(endJob)
	withLimit := clientv3.WithLimit(kvRangeLimit)

	for {
		key := etcdutil.JobPath(kv.path, nextID)
		resp, err := kv.BatchLoadWithOpts(key, withRange, withLimit)
		if err != nil {
			return errors.Trace(err)
		}

		for _, item := range resp.Kvs {
			job := &whpb.Job{}
			if err := job.Unmarshal(item.Value); err != nil {
				return errors.Trace(err)
			}

			nextID = job.GetId() + 1
			jobs.setJob(newJobInfo(job))
		}

		if len(resp.Kvs) < int(kvRangeLimit) {
			return nil
		}
	}
}

func (kv *kv) loadWorkers(workers *workersInfo) error {
	nextID := uint64(0)
	endWorker := etcdutil.WorkerPath(kv.path, math.MaxUint64)
	withRange := clientv3.WithRange(endWorker)
	withLimit := clientv3.WithLimit(kvRangeLimit)

	for {
		key := etcdutil.WorkerPath(kv.path, nextID)
		resp, err := kv.BatchLoadWithOpts(key, withRange, withLimit)
		if err != nil {
			return errors.Trace(err)
		}

		for _, item := range resp.Kvs {
			worker := &whpb.Worker{}
			if err := worker.Unmarshal(item.Value); err != nil {
				return errors.Trace(err)
			}

			nextID = worker.GetId() + 1
			workers.setWorker(newWorkerInfo(worker))
		}

		if len(resp.Kvs) < int(kvRangeLimit) {
			return nil
		}
	}
}

// returns map[workerID][]jobID, error
func (kv *kv) loadContracts() (map[uint64][]uint64, error) {
	nextJobID := uint64(0)
	nextWorkerID := uint64(0)
	endContract := etcdutil.ContractPath(kv.path, math.MaxUint64, math.MaxUint64)
	withRange := clientv3.WithRange(endContract)
	withLimit := clientv3.WithLimit(kvRangeLimit)
	contracts := make(map[uint64][]uint64)

	for {
		key := etcdutil.ContractPath(kv.path, nextJobID, nextWorkerID)
		resp, err := kv.BatchLoadWithOpts(key, withRange, withLimit)
		if err != nil {
			return contracts, errors.Trace(err)
		}

		for _, item := range resp.Kvs {
			jobID, workerID, err := parseContractKey(item)
			if err != nil {
				return contracts, errors.Trace(err)
			}
			nextWorkerID = workerID + 1
			nextJobID = jobID
			contracts[workerID] = append(contracts[workerID], jobID)
		}

		if len(resp.Kvs) < int(kvRangeLimit) {
			return contracts, nil
		}
	}
}

func (kv *kv) loadPipelines() (map[uint64][]byte, error) {
	resp, err := kv.BatchLoadWithOpts(etcdutil.PipelinePathPrefix(kv.path), clientv3.WithPrefix())
	if err != nil {
		return nil, errors.Trace(err)
	}

	rawPipelines := make(map[uint64][]byte, len(resp.Kvs))
	for _, item := range resp.Kvs {
		jobID, err := parsePipelineKey(item)
		if err != nil {
			return rawPipelines, errors.Trace(err)
		}
		rawPipelines[jobID] = item.Value
	}

	return rawPipelines, nil
}

func (kv *kv) loadFence(fenceID uint64, fence *whpb.Fence) (bool, error) {
	return kv.LoadProto(etcdutil.FencePath(kv.path, fenceID), fence)
}

func (kv *kv) loadWorker(workerID uint64, worker *whpb.Worker) (bool, error) {
	return kv.LoadProto(etcdutil.WorkerPath(kv.path, workerID), worker)
}

func (kv *kv) loadPipeline(jobID uint64) ([]byte, error) {
	return kv.Load(etcdutil.PipelinePath(kv.path, jobID))
}

func (kv *kv) saveJob(job *whpb.Job) error {
	return kv.SaveProto(etcdutil.JobPath(kv.path, job.GetId()), job)
}

func (kv *kv) saveWorker(worker *whpb.Worker) error {
	return kv.SaveProto(etcdutil.WorkerPath(kv.path, worker.GetId()), worker)
}

func (kv *kv) saveFence(fence *whpb.Fence) error {
	return kv.SaveProto(etcdutil.FencePath(kv.path, fence.GetId()), fence)
}

func (kv *kv) saveContract(jobID, workerID uint64, lease int64) error {
	lcr, err := kv.GrantLease(lease)
	if err != nil {
		return errors.Trace(err)
	}

	opts := []clientv3.OpOption{clientv3.WithLease(lcr.ID)}
	return kv.Save(etcdutil.ContractPath(kv.path, jobID, workerID), string(uint64ToBytes(jobID)), opts...)
}

func (kv *kv) savePipeline(jobID uint64, info []byte) error {
	return kv.Save(etcdutil.PipelinePath(kv.path, jobID), string(info))
}

func (kv *kv) deletePipeline(jobID uint64) error {
	return kv.Delete(etcdutil.PipelinePath(kv.path, jobID))
}

func (kv *kv) deleteContract(jobID, workerID uint64) error {
	return kv.Delete(etcdutil.ContractPath(kv.path, jobID, workerID))
}

func (kv *kv) loadSchemaInfo(host string, port int32, schemaInfo *whpb.SchemaInfo) (bool, error) {
	return kv.LoadProto(etcdutil.SchemaMetaPath(kv.path, host, port), schemaInfo)
}

func (kv *kv) deleteSchemaInfo(host string, port int32) error {
	return kv.Delete(etcdutil.SchemaMetaPath(kv.path, host, port))
}