package ether

import (
	"strconv"
	"sync"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/whpb"
)

const (
	// MaxScore is maximum score
	MaxScore = 10
	// MaxOfflineThreshold is the max offline time , when this time is exceeded, be judged offline
	MaxOfflineThreshold = 300
	// MaxSaveThreshold is the max save time , determines whether the state needs to be saved
	MaxSaveThreshold = 120
)

// clusterInfo is cluster cache layer.
type clusterInfo struct {
	sync.RWMutex

	id        IDAllocator
	clusterID uint64
	kv        *kv

	jobs    *jobsInfo
	workers *workersInfo
	fences  *fencesInfo

	// cache front job
	pipelines map[uint64][]byte
}

func loadClusterInfo(clusterID uint64, kv *kv, id IDAllocator) (*clusterInfo, error) {
	c := &clusterInfo{
		id:        id,
		clusterID: clusterID,
		kv:        kv,
		jobs:      newJobsInfo(),
		workers:   newWorkersInfo(),
		fences:    newFencesInfo(),
		pipelines: make(map[uint64][]byte, 128),
	}

	if err := c.loadResources(); err != nil {
		return nil, errors.Trace(err)
	}
	return c, nil
}

// only load jobs and workers
// load contracts while start controller
// load fences while need
func (c *clusterInfo) loadResources() error {
	c.Lock()
	defer c.Unlock()

	if err := c.loadJobs(); err != nil {
		return errors.Trace(err)
	}
	if err := c.loadWorkers(); err != nil {
		return errors.Trace(err)
	}
	if err := c.loadPipelines(); err != nil {
		return errors.Trace(err)
	}

	return nil
}

/************************ contracts ************************/

func (c *clusterInfo) createContract(jobID, workerID uint64, lease int64) error {
	log.Infof("[ether] create contract job %d vs worker %d", jobID, workerID)
	c.Lock()
	defer c.Unlock()

	job := c.jobs.getJob(jobID)
	if job == nil || !job.canRunning() {
		return errors.Errorf("job %d is not exist or not running", jobID)
	}

	log.Infof("create contract job %d vs worker %d", jobID, workerID)
	err := c.kv.saveContract(jobID, workerID, lease)
	if err != nil {
		return errors.Trace(err)
	}

	c.workers.signContract(workerID, jobID)
	c.jobs.signContract(jobID, workerID)
	return nil
}

func (c *clusterInfo) terminateContract(jobID, workerID uint64) {
	c.Lock()
	c.workers.terminateContract(workerID, jobID)
	c.jobs.terminateContract(jobID, workerID)
	// FIXME: we check error clearly
	c.kv.deleteContract(jobID, workerID)
	c.Unlock()
}

// contracts
func (c *clusterInfo) loadContracts() error {
	c.Lock()
	defer c.Unlock()
	contracts, err := c.kv.loadContracts()
	if err != nil {
		return errors.Trace(err)
	}

	for workerID, jobIDs := range contracts {
		for _, jobID := range jobIDs {
			c.workers.signContract(workerID, jobID)
			c.jobs.signContract(jobID, workerID)
		}
	}
	return nil
}

/************************ job ************************/

func (c *clusterInfo) storeJob(job *whpb.Job) error {
	c.Lock()
	defer c.Unlock()
	err := c.kv.saveJob(job)
	if err != nil {
		return errors.Trace(err)
	}
	c.jobs.setJob(newJobInfo(job))
	return nil
}

func (c *clusterInfo) findPortalJob(host string, port int32) *jobInfo {
	c.RLock()
	job := c.jobs.findPortalJob(host, port)
	c.RUnlock()
	return job
}

// getJob returns specified job by id
func (c *clusterInfo) getJob(jobID uint64) *jobInfo {
	c.RLock()
	job := c.jobs.getJob(jobID)
	c.RUnlock()
	return job
}

func (c *clusterInfo) canSchedule(jobID uint64) bool {
	c.RLock()
	defer c.RUnlock()

	return c.jobs.canSchedule(jobID)
}

func (c *clusterInfo) adjustReplicate(jobID uint64, replicate int32) error {
	c.Lock()
	defer c.Unlock()

	job := c.jobs.adjustReplicate(jobID, replicate)
	if job == nil {
		return errors.NotFoundf("[ether] job %d", jobID)
	}

	return c.kv.saveJob(job.getJob())
}

func (c *clusterInfo) pauseJob(jobID uint64) error {
	c.Lock()
	defer c.Unlock()
	job, err := c.jobs.pauseJob(jobID)
	if err != nil {
		return errors.Trace(err)
	}
	return c.kv.saveJob(job.Job)
}

func (c *clusterInfo) restartJob(jobID uint64) error {
	c.Lock()
	defer c.Unlock()
	job, err := c.jobs.restartJob(jobID)
	if err != nil {
		return errors.Trace(err)
	}
	return errors.Trace(c.kv.saveJob(job.Job))
}

func (c *clusterInfo) deleteJob(jobID uint64) error {
	c.Lock()
	defer c.Unlock()

	job := c.jobs.deleteJob(jobID)
	if job == nil {
		return errors.NotFoundf("[ether] job %d", jobID)
	}
	return errors.Trace(c.kv.saveJob(job.getJob()))
}

func (c *clusterInfo) loadJobs() error {
	start := time.Now()
	if err := c.kv.loadJobs(c.jobs); err != nil {
		return errors.Trace(err)
	}
	log.Infof("[ether] get all jobs %d cost %v", c.jobs.count(), time.Since(start))
	return nil
}

func (c *clusterInfo) allJobs() []*jobInfo {
	c.RLock()
	jobs := c.jobs.all()
	c.RUnlock()
	return jobs
}

/************************ fence ************************/

// createFence creates fence by given topics
func (c *clusterInfo) createFence(topics []*whpb.Topic) (*whpb.Fence, error) {
	// get all partitions
	bitmaps := make([]*whpb.TopicPartitionBitmap, 0, len(topics))

	for _, topic := range topics {
		bitmaps = append(bitmaps, &whpb.TopicPartitionBitmap{
			Topic:                 topic,
			PartitionBitmap:       partitionsBitmap(topic.GetPartitionNum()),
			PartitionMidwayBitmap: make([]uint64, int((topic.GetPartitionNum()+63)/64)),
		})
	}

	id, err := c.id.Alloc()
	if err != nil {
		log.Errorf("[ether] alloc id error %v", err)
		return nil, errors.Errorf("alloc fence id error %v", err)
	}

	// save fence
	fence := &whpb.Fence{
		Id:      id,
		Bitmaps: bitmaps,
	}
	c.Lock()
	err = c.kv.saveFence(fence)
	if err != nil {
		c.Unlock()
		return nil, errors.Annotatef(err, "save fence %d in etcd", id)
	}
	c.fences.setFence(fence)

	c.Unlock()
	return fence, nil
}

// completePartitions fill partitions bit in fence bitmap by give partition IDs
func (c *clusterInfo) completePartitions(fenceID uint64, jobID uint64, topic *whpb.Topic, partitionIDs []uint64, final bool) error {
	c.Lock()
	defer c.Unlock()

	fence := c.fences.getJobFence(fenceID, jobID)
	if fence == nil {
		fence = &whpb.Fence{}
		ok, err := c.kv.loadFence(fenceID, fence)
		if err != nil {
			return errors.Trace(err)
		}
		if !ok {
			return errors.NotFoundf("fence %d", fenceID)
		}
	}

	// mark finished partitions in bitmap
	for i, topicPartitionBitmap := range fence.GetBitmaps() {
		if topic.GetName() == topicPartitionBitmap.GetTopic().GetName() {
			// set partitions bit map
			bitmap := topicPartitionBitmap.GetPartitionBitmap()
			midwayBitmap := topicPartitionBitmap.GetPartitionMidwayBitmap()
			for _, pid := range partitionIDs {
				bindex := int(pid / 64)
				if bindex >= len(bitmap) {
					return errors.Errorf("[ether] fenceid %d, topic %s, wrong partitionIDs %v, fence bitmap it %v", fenceID, topic.GetName(), partitionIDs, bitmap)
				}
				bitmap[pid/64] = (bitmap[pid/64]) | (1 << (pid % 64)) ^ (1 << (pid % 64))
				if final {
					midwayBitmap[pid/64] = (midwayBitmap[pid/64]) | (1 << (pid % 64)) ^ (1 << (pid % 64))
				} else {
					midwayBitmap[pid/64] = (midwayBitmap[pid/64]) | (1 << (pid % 64))
				}
			}
			topicPartitionBitmap.PartitionBitmap = bitmap
			topicPartitionBitmap.PartitionMidwayBitmap = midwayBitmap
			fence.Bitmaps[i] = topicPartitionBitmap
			break
		}
	}

	// save in cache
	c.fences.setJobFence(fence, jobID)
	return nil
}

// isFenceSynced return is synced in midway or final
func (c *clusterInfo) isFenceSynced(fenceID uint64, jobID uint64) (bool, bool) {
	c.RLock()
	midway, final := c.fences.isSynced(fenceID, jobID)
	c.RUnlock()
	return midway, final
}

/************************ worker ************************/

// createWorker register new worker.
func (c *clusterInfo) createWorker(worker *whpb.Worker) error {
	var err error
	worker.Id, err = c.id.Alloc()
	if err != nil {
		return err
	}
	log.Infof("[ether] create new worker %+v", worker)
	worker.CreateTime = time.Now().Unix()
	groupVal := strconv.FormatUint(worker.Id, 10)
	if worker.Labels == nil {
		worker.Labels = make(map[string]*whpb.Label)
	}
	worker.Labels["group"] = &whpb.Label{Vals: []string{groupVal}}
	worker.Labels["type"] = &whpb.Label{Vals: []string{worker.Type.String()}}

	return c.storeWorker(worker)
}

// getWorker returns specified worker by id
func (c *clusterInfo) getWorker(workerID uint64) *workerInfo {
	c.RLock()
	worker := c.workers.getWorker(workerID)
	c.RUnlock()
	return worker
}

func (c *clusterInfo) storeWorker(worker *whpb.Worker) error {
	c.Lock()
	worker.LastTime = time.Now().Unix()
	err := c.kv.saveWorker(worker)
	if err != nil {
		c.Unlock()
		return errors.Trace(err)
	}
	c.workers.setWorker(newWorkerInfo(worker))
	c.Unlock()
	return nil
}

// renewContract renews contract
func (c *clusterInfo) renewContract(jobID, workerID uint64, lease int64) (isExpired bool, err error) {
	c.RLock()
	defer c.RUnlock()

	ok := c.workers.existContract(workerID, jobID)
	if !ok {
		return true, nil
	}

	// check job status
	job := c.jobs.getJob(jobID)
	if job == nil || !job.canRunning() {
		return true, nil
	}

	// save contract
	err = c.kv.saveContract(jobID, workerID, lease)
	return false, errors.Trace(err)
}

func (c *clusterInfo) handleWorkerHeartbeat(workerID uint64, jobStatus []*whpb.JobStatus) error {
	c.Lock()
	defer c.Unlock()
	worker := c.workers.getWorker(workerID)
	if worker == nil {
		return errors.NotFoundf("worker %d", workerID)
	}

	// fixme: reduce saveWorker
	worker.LastTime = time.Now().Unix()
	err := c.kv.saveWorker(worker.Worker)
	if err != nil {
		return errors.Trace(err)
	}
	c.workers.setWorker(worker)

	// renew worker and store job status
	for _, status := range jobStatus {
		// check contract
		// in normal case we wouldn't meet it:
		// before contract expired, worker would give up the contract
		ok := worker.existContract(status.GetId())
		if !ok {
			log.Errorf("[ether] worker %d dont't have job %d", worker.GetId(), status.GetId())
			continue
		}

		// store job status
		job, needSave := c.jobs.updateStatus(workerID, status)
		if job == nil {
			log.Errorf("[ether] job %d not found", status.GetId())
			continue
		}

		// fixme: save in a batch
		if needSave {
			err = c.kv.saveJob(job.Job)
			if err != nil {
				log.Errorf("[ether] save job status errpr %v", err)
			}
		}
	}
	return nil
}

func (c *clusterInfo) loadWorkers() error {
	start := time.Now()
	if err := c.kv.loadWorkers(c.workers); err != nil {
		return errors.Trace(err)
	}
	log.Infof("[ether] get all workers %d cost %v", c.workers.count(), time.Since(start))
	return nil
}

func (c *clusterInfo) allWorkers() []*workerInfo {
	c.RLock()
	workers := c.workers.all()
	c.RUnlock()
	return workers
}

/************************ pipeline ************************/

// getWorker returns specified worker by id
func (c *clusterInfo) getPipeline(jobID uint64) []byte {
	c.RLock()
	pipeline := c.pipelines[jobID]
	c.RUnlock()
	return pipeline
}

func (c *clusterInfo) storePipeline(jobID uint64, rawPipeline []byte) error {
	c.Lock()
	err := c.kv.savePipeline(jobID, rawPipeline)
	if err != nil {
		c.Unlock()
		return errors.Trace(err)
	}
	c.pipelines[jobID] = rawPipeline
	c.Unlock()
	return nil
}

func (c *clusterInfo) deletePipeline(jobID uint64) error {
	c.Lock()
	err := c.kv.deletePipeline(jobID)
	if err != nil {
		c.Unlock()
		return errors.Trace(err)
	}
	delete(c.pipelines, jobID)
	c.Unlock()
	return nil
}

func (c *clusterInfo) allPipelines() map[uint64][]byte {
	c.RLock()
	pipelines := make(map[uint64][]byte, len(c.pipelines))
	for jobID, info := range c.pipelines {
		pipelines[jobID] = info
	}
	c.RUnlock()
	return pipelines
}

func (c *clusterInfo) loadPipelines() error {
	start := time.Now()
	rawPipelines, err := c.kv.loadPipelines()
	if err != nil {
		return errors.Trace(err)
	}
	log.Infof("[ether] get all workers %d cost %v", c.workers.count(), time.Since(start))

	c.pipelines = rawPipelines
	return nil
}