package framework

import (
	"sync"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/config"
	"github.com/pingcap/wormhole/pkg/ether-client"
	"github.com/pingcap/wormhole/pkg/mq/zk"
	"github.com/pingcap/wormhole/pkg/whpb"
	"github.com/pingcap/wormhole/portal/mysql"
	stargate "github.com/pingcap/wormhole/stargate/mysql"
)

// Handler contains configuration options for handler
type Handler struct {
	// remember to update job status.
	job       *whpb.Job
	processor Processor
	cfg       *config.Config
	ec        ether.Client

	zkClient *zk.Client
	wg       sync.WaitGroup
}

// NewHandler creates a handler
func NewHandler(cfg *config.Config, ec ether.Client, zkClient *zk.Client) *Handler {
	h := &Handler{}
	h.cfg = cfg
	h.ec = ec
	h.zkClient = zkClient
	return h
}

// AssignJob assigns a job to the handler.
func (h *Handler) AssignJob(job *whpb.Job) error {
	log.Infof("[worker] assign job. job_id: %d, status: %+v, source: %+v, type: %+v, label: %+v, annotations %+v, dependencies %+v",
		job.Id, job.Status, job.Source, job.Type, job.Labels, job.Annotations, job.Dependencies)

	if isUnacceptableJob(job) {
		return errors.Errorf("job %d unacceptable because of state %v", job.Id, job.Status.State)
	}

	//reset job message
	job.Status.Message = ""
	h.job = job

	var err error
	if job.GetType() == whpb.WorkerType_Portal && job.GetSource() == whpb.SourceType_mysql_binlog {
		h.processor, err = mysql.NewSyncer(job, h.ec, h.zkClient, h.cfg)
	} else if job.GetType() == whpb.WorkerType_Portal && job.GetSource() == whpb.SourceType_mysql_dump {
		h.processor, err = mysql.NewDumper(job, h.ec, h.zkClient, h.cfg)
	} else if job.GetType() == whpb.WorkerType_Stargate && (job.GetSource() == whpb.SourceType_mysql_dump || job.GetSource() == whpb.SourceType_mysql_binlog) {
		h.processor, err = stargate.New(job, h.ec, h.zkClient, h.cfg)
	}
	if err != nil {
		return errors.Trace(err)
	}

	go func() {
		h.wg.Add(1)
		log.Infof("[worker] job %+v begin processing", h.job)
		err := h.processor.Process()
		if err != nil {
			log.Errorf("[worker] process error: %v", errors.ErrorStack(err))
			h.job.Status.State = whpb.JobState_Abnormal
			h.job.Status.Message = err.Error()
		} else {
			h.job.Status.Stage = whpb.JobStage_Resolved
		}
		h.wg.Done()
		log.Infof("[worker] job %+v stop processing", h.job)
	}()

	return nil
}

func isUnacceptableJob(job *whpb.Job) bool {
	if job.Status.State != whpb.JobState_Normal {
		return true
	}

	if job.Status.Stage == whpb.JobStage_New || job.Status.Stage == whpb.JobStage_Initializing || job.Status.Stage == whpb.JobStage_Running {
		return false
	}

	return true
}

// GetStatus return job status of the Processor
func (h *Handler) GetStatus() *whpb.JobStatus {
	if h.processor != nil {
		h.job.Status.Total = h.processor.GetStatus().Total
		h.job.Status.Pos = h.processor.GetStatus().Pos
		h.job.Status.Qps = h.processor.GetStatus().Qps
		h.job.Status.Percent = h.processor.GetStatus().Percent
	}
	return h.job.Status
}

// Close closes the handler.
func (h *Handler) Close() {
	if h.processor != nil {
		h.processor.Close()
	}
	h.wg.Wait()
	log.Infof("[worker] handler closed with job_id %d, source_type %v, worker_type %v, %+v ",
		h.job.GetId(), h.job.GetSource(), h.job.GetType(), h.job)
}