package ether

import (
	"time"

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

type jobInfo struct {
	contracts map[uint64]*whpb.JobStatus
	*whpb.Job
	lastSaveTime int64
}

func newJobInfo(job *whpb.Job) *jobInfo {
	return &jobInfo{
		contracts:    make(map[uint64]*whpb.JobStatus),
		Job:          job,
		lastSaveTime: time.Now().Unix(),
	}
}

func (j *jobInfo) getJob() *whpb.Job {
	return j.Job
}

func (j *jobInfo) isPortalJob() bool {
	return j.GetType() == whpb.WorkerType_Portal
}

func (j *jobInfo) isStargateJob() bool {
	return j.GetType() == whpb.WorkerType_Stargate
}

func (j *jobInfo) isSamePortalJob(host string, port int32) bool {
	if j.isDeleted() {
		return false
	}
	if !j.isPortalJob() {
		return false
	}
	switch j.GetSource() {
	case whpb.SourceType_mysql_binlog:
		jobInfo := &whpb.MysqlBinlogJob{}
		err := jobInfo.Unmarshal(j.GetInfo())
		if err != nil {
			log.Infof("[ether] unmarshal job error %v", err)
			return false
		}

		db := jobInfo.GetSource()
		return db.GetHost() == host && db.GetPort() == port
	}

	return false
}

func (j *jobInfo) adjustReplicate(replicate int32) {
	j.Job.Status.Replicate = replicate
}

func (j *jobInfo) isNew() bool {
	return j.Job.Status.GetStage() == whpb.JobStage_New
}

func (j *jobInfo) isInitializing() bool {
	return j.Job.Status.GetStage() == whpb.JobStage_Initializing
}

func (j *jobInfo) beInitializing() {
	j.Job.Status.Stage = whpb.JobStage_Initializing
}

func (j *jobInfo) isRunning() bool {
	return j.Job.Status.GetStage() == whpb.JobStage_Running
}

func (j *jobInfo) beRunning() {
	j.Job.Status.Stage = whpb.JobStage_Running
}

func (j *jobInfo) isResolved() bool {
	return j.Job.Status.GetStage() == whpb.JobStage_Resolved
}

func (j *jobInfo) beResolved() {
	j.Job.Status.Stage = whpb.JobStage_Resolved
}

func (j *jobInfo) isDeleting() bool {
	return j.Job.Status.GetStage() == whpb.JobStage_Deleting
}

func (j *jobInfo) beDeleting() {
	j.Job.Status.Stage = whpb.JobStage_Deleting
}

func (j *jobInfo) isDeleted() bool {
	return j.Job.Status.GetStage() == whpb.JobStage_Deleted
}

func (j *jobInfo) beDeleted() {
	j.Job.Status.Stage = whpb.JobStage_Deleted
}

// state
func (j *jobInfo) isNormalState() bool {
	return j.Job.Status.GetState() == whpb.JobState_Normal
}

func (j *jobInfo) beNormalState() {
	j.Job.Status.State = whpb.JobState_Normal
}

func (j *jobInfo) isAbnormalState() bool {
	return j.Job.Status.GetState() == whpb.JobState_Abnormal
}

func (j *jobInfo) beAbnormalState(message string) {
	j.Job.Status.State = whpb.JobState_Abnormal
	j.Job.Status.Message = message
}

func (j *jobInfo) isPausedState() bool {
	return j.Job.Status.GetState() == whpb.JobState_Paused
}

func (j *jobInfo) bePausedState() {
	j.Job.Status.State = whpb.JobState_Paused
}

func (j *jobInfo) restart() {
	j.beNormalState()
}

func (j *jobInfo) canRunning() bool {
	return j.isNormalState() && (j.isNew() || j.isInitializing() || j.isRunning())
}

func (j *jobInfo) canScheduled() bool {
	if j.isAbnormalState() {
		return false
	}

	if j.isNew() || j.isInitializing() {
		return j.isNormalState() && len(j.contracts) == 0
	}

	return j.isRunning() && int(j.Job.Status.GetReplicate()) > len(j.contracts)
}

func (j *jobInfo) updateStatus(workerID uint64, jobStatus *whpb.JobStatus) bool {
	j.contracts[workerID] = jobStatus
	needSave := j.transferStage(jobStatus)

	//  report abnormal while meet error
	if jobStatus.State == whpb.JobState_Abnormal {
		j.beAbnormalState(jobStatus.Message)
		needSave = true
	}

	var (
		total  uint64
		finish uint64
		qps    uint64
		pos    *whpb.Position
	)
	// FIXME iterate contracts?
	for _, status := range j.contracts {
		total += status.GetTotal()
		finish += status.GetTotal() * status.GetPercent()
		qps += status.GetQps()
		if pos == nil {
			pos = status.GetPos()
			continue
		}
		res, err := comparePosition(pos, status.GetPos(), j.GetSource())
		if err != nil {
			log.Errorf("[ether] compare position error %v, pos %+v vs %+v", err, pos, status.GetPos())
			continue
		}
		if res > 1 {
			pos = status.GetPos()
		}
	}
	j.Status.Total = total
	if total != 0 {
		j.Status.Percent = finish / total
	}
	j.Status.Qps = qps
	j.Status.Pos = pos

	if j.needSave() {
		needSave = true
	}
	return needSave
}

func (j *jobInfo) transferStage(status *whpb.JobStatus) bool {
	switch status.Stage {
	case whpb.JobStage_Initializing:
		if j.isNew() {
			j.beInitializing()
			return true
		}
	case whpb.JobStage_Running:
		if j.isNew() || j.isInitializing() {
			j.beRunning()
			return true
		}
	case whpb.JobStage_Resolved:
		if j.isNew() || j.isInitializing() || j.isRunning() {
			for _, status := range j.contracts {
				if status.GetStage() != whpb.JobStage_Resolved {
					return false
				}
			}
			j.beResolved()
			return true
		}
	}

	return false
}

func (j *jobInfo) needSave() bool {
	now := time.Now().Unix()
	if now-j.lastSaveTime > MaxSaveThreshold {
		j.lastSaveTime = now
		return true
	}

	return false
}

func (j *jobInfo) signContract(workerID uint64) {
	j.contracts[workerID] = proto.Clone(j.GetStatus()).(*whpb.JobStatus)
}

func (j *jobInfo) terminateContract(workerID uint64) {
	delete(j.contracts, workerID)
}

func (j *jobInfo) clone() *jobInfo {
	worker := make(map[uint64]*whpb.JobStatus)
	for workerID, status := range j.contracts {
		worker[workerID] = proto.Clone(status).(*whpb.JobStatus)
	}

	return &jobInfo{
		Job:          proto.Clone(j.Job).(*whpb.Job),
		contracts:    worker,
		lastSaveTime: j.lastSaveTime,
	}
}

func (j *jobInfo) GetStatusScore() int {
	if j.canRunning() {
		return MaxScore
	}
	return 0
}

func (j *jobInfo) GetContracts() []uint64 {
	contracts := make([]uint64, len(j.contracts))
	index := 0
	for id := range j.contracts {
		contracts[index] = id
		index++
	}
	return contracts
}

// jobsInfo contains all job information.
type jobsInfo struct {
	jobs map[uint64]*jobInfo
}

func newJobsInfo() *jobsInfo {
	return &jobsInfo{
		jobs: make(map[uint64]*jobInfo),
	}
}

func (j *jobsInfo) setJob(job *jobInfo) {
	j.jobs[job.Id] = job
}

func (j *jobsInfo) getJob(id uint64) *jobInfo {
	job, ok := j.jobs[id]
	if !ok {
		return nil
	}

	return job.clone()
}

func (j *jobsInfo) findPortalJob(host string, port int32) *jobInfo {
	for _, job := range j.jobs {
		if job.isSamePortalJob(host, port) {
			return job.clone()
		}
	}
	return nil
}

func (j *jobsInfo) canSchedule(jobID uint64) bool {
	job, ok := j.jobs[jobID]
	if !ok {
		return false
	}

	dependencies := job.GetDependencies()
	for _, id := range dependencies {
		dependentJob, ok := j.jobs[id]
		if !ok {
			continue
		}
		if !dependentJob.isResolved() {
			return false
		}
	}

	return job.canScheduled()
}

func (j *jobsInfo) updateStatus(workerID uint64, status *whpb.JobStatus) (*jobInfo, bool) {
	job, ok := j.jobs[status.GetId()]
	if !ok {
		return nil, false
	}

	log.Debugf("[ether] job status %+v", status)
	needSave := job.updateStatus(workerID, status)

	return job.clone(), needSave
}

func (j *jobsInfo) signContract(jobID uint64, workerID uint64) {
	job, ok := j.jobs[jobID]
	if !ok {
		return
	}
	job.signContract(workerID)
}

func (j *jobsInfo) terminateContract(jobID uint64, workerID uint64) *jobInfo {
	job, ok := j.jobs[jobID]
	if !ok {
		return nil
	}
	job.terminateContract(workerID)
	return job.clone()
}

func (j *jobsInfo) restartJob(jobID uint64) (*jobInfo, error) {
	job, ok := j.jobs[jobID]
	if !ok {
		return nil, errors.NotFoundf("job %d", jobID)
	}
	if job.isDeleted() {
		return nil, errors.Errorf("[ether] job %d was deleted", jobID)
	}

	job.restart()
	return job.clone(), nil
}

func (j *jobsInfo) pauseJob(jobID uint64) (*jobInfo, error) {
	job, ok := j.jobs[jobID]
	if !ok {
		return nil, errors.NotFoundf("job %d", jobID)
	}
	if !job.isRunning() || job.isAbnormalState() {
		return nil, errors.Errorf("[ether] can't pause non-runing or abnormal job %d [stage: %s, state %s]", jobID, job.Status.GetStage(), job.Status.GetState())
	}

	job.bePausedState()
	return job.clone(), nil
}

func (j *jobsInfo) deleteJob(jobID uint64) *jobInfo {
	job, ok := j.jobs[jobID]
	if !ok {
		return nil
	}
	job.beDeleted()
	return job.clone()
}

func (j *jobsInfo) adjustReplicate(jobID uint64, replicate int32) *jobInfo {
	job, ok := j.jobs[jobID]
	if !ok {
		return nil
	}
	job.adjustReplicate(replicate)
	return job.clone()
}

func (j *jobsInfo) count() int {
	return len(j.jobs)
}

func (j *jobsInfo) all() []*jobInfo {
	jobs := make([]*jobInfo, len(j.jobs))
	index := 0
	for _, job := range j.jobs {
		jobs[index] = job.clone()
		index++
	}

	return jobs
}