package api

import (
	"encoding/json"
	"net/http"
	"strconv"
	"strings"

	"github.com/gorilla/mux"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/ether"
	"github.com/pingcap/wormhole/pkg/whpb"
	"github.com/unrolled/render"
)

const splitJobFactor = 32

// annotations key for job
const (
	dsNameTag     = "dsName"
	jobTypeTag    = "jobTypeForFront"
	mainJobTag    = "mainJobID"
	createTimeTag = "createTime"
)

const (
	portalDump     = "portal_dump"
	portalBinlog   = "portal_binlog"
	stargateDump   = "stargate_dump"
	stargateBinlog = "stargate_binlog"
)

type jobHandler struct {
	*ether.Handler
	r *render.Render
}

func newJobHandler(handler *ether.Handler, r *render.Render) *jobHandler {
	return &jobHandler{
		Handler: handler,
		r:       r,
	}
}

// Datasource presents data source, corresponds to a job in ether
type Datasource struct {
	ID           uint64          `json:"id"`
	Name         string          `json:"name"`
	Type         string          `json:"type"`
	DB           whpb.DBAccount  `json:"db"`
	PartitionNum int32           `json:"partition_num"`
	Status       *whpb.JobStatus `json:"status"`
	CreateTime   string          `json:"create_time"`
	ServerID     int32           `json:"server_id"`
}

// use binlog job represent a datasource
// there is no actually datasource info stored in etcd
func (h *jobHandler) CreateDataSource(w http.ResponseWriter, r *http.Request) {
	source := &Datasource{}
	err := readJSON(r.Body, source)
	if err != nil {
		h.r.JSON(w, http.StatusBadRequest, err.Error())
		return
	}
	log.Infof("[ether] create datasource %+v", source)

	// check, validate param + distinct config
	err = validateDB(source.DB)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] wrong DB config %v", err))
		return
	}
	// validate the serverid, if it is 0, use a rand number
	source.ServerID = validateServerID(source.ServerID)

	// TODO, refactor datasoure identifier, now are host+port
	existJob, err := h.FindSameDataSourceJob(&source.DB)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] validate same data source %v", err))
		return
	}
	if existJob != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] have same data source"))
		return
	}

	var (
		jobInfo  []byte
		sourceTp whpb.SourceType
	)
	switch source.Type {
	case "mysql":
		sourceTp = whpb.SourceType_mysql_binlog
		info := &whpb.MysqlBinlogJob{
			Source:       source.DB,
			PartitionNum: source.PartitionNum,
			ServerId:     source.ServerID,
			Schemas:      nil,
		}
		jobInfo, err = info.Marshal()
		if err != nil {
			log.Errorf("[ether] marshal error %v", errors.ErrorStack(err))
			h.r.JSON(w, http.StatusInternalServerError, err.Error())
			return
		}
	default:
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] unsupport source type %v", source.Type))
		return
	}

	labels := make(map[string]*whpb.Label)
	labels["type"] = &whpb.Label{Vals: []string{whpb.WorkerType_Portal.String()}}

	id, err := h.AllocID()
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] accocate id error %v", err))
		return
	}

	job := &whpb.Job{
		Id:          id,
		Type:        whpb.WorkerType_Portal,
		Source:      sourceTp,
		Info:        jobInfo,
		Annotations: map[string]string{dsNameTag: source.Name, jobTypeTag: "dataSource", createTimeTag: source.CreateTime},
		Labels:      labels,
		Status: &whpb.JobStatus{
			Id:        id,
			Type:      portalBinlog,
			Replicate: 1,
		},
	}
	err = h.StoreJob(job)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] store job error %v", err))
		return
	}
	// use job id as datasource id
	source.ID = job.GetId()
	h.r.JSON(w, http.StatusOK, successResponse(source))
}

func (h *jobHandler) GetDataSources(w http.ResponseWriter, r *http.Request) {
	jobs, err := h.GetDataSourceJobs()
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] get datasource jobs error %v", err))
		return
	}

	ds, err := h.formatDataSources(jobs)
	if err != nil {
		log.Errorf("[ether] format data source error %v", errors.ErrorStack(err))
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}
	h.r.JSON(w, http.StatusOK, successResponse(ds))
}

func (h *jobHandler) GetDataSource(w http.ResponseWriter, r *http.Request) {
	// fetch datasource id, actually used as jobid for find binlog job
	id := mux.Vars(r)["id"]
	jobID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse data_source_id %s error %v", id, err))
		return
	}

	job, err := h.GetJob(jobID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] get job %d error %v", jobID, err))
		return
	}
	ds, err := h.formatDataSources([]*whpb.Job{job})
	if err != nil {
		log.Errorf("[ether] format data source error %v", errors.ErrorStack(err))
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}

	log.Debugf("[ether]  get datasource %+v", ds[0])
	h.r.JSON(w, http.StatusOK, successResponse(ds[0]))
}

func (h *jobHandler) DeleteDatasource(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	jobID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse datasource id %s error %v", id, err))
		return
	}

	log.Infof("[ether] delete datasource %d", jobID)
	job, err := h.GetJob(jobID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] get job %d error %v", jobID, err))
		return
	}
	ds, err := h.formatDataSources([]*whpb.Job{job})
	if err != nil {
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}

	var ok bool
	if len(ds) > 0 {
		err = h.DeleteSchemaInfo(ds[0].DB.GetHost(), ds[0].DB.GetPort())
		if err != nil {
			h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to delete schemainfo error %v", err))
			return
		}

		ok, err = h.tryDeleteDatasource(&ds[0].DB)
		if err != nil {
			log.Errorf("try to delete datasource error %v", err)
		}
		if !ok {
			h.r.JSON(w, http.StatusOK, errResponsef("fail to delete datasource, other pipelines use it"))
			return
		}
	} else {
		err = h.DeleteJob(jobID)
		if err != nil {
			h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to delete datasource error %v", err))
			return
		}
	}

	h.r.JSON(w, http.StatusOK, successResponse(ds[0]))
}

func (h *jobHandler) formatDataSources(jobs []*whpb.Job) ([]*Datasource, error) {
	var (
		dataSources []*Datasource
		deletedDS   []uint64
	)
	for _, job := range jobs {
		if job.GetStatus().GetStage() == whpb.JobStage_Deleted {
			deletedDS = append(deletedDS, job.GetId())
			continue
		}

		ds := &Datasource{
			ID:         job.GetId(),
			Name:       job.GetAnnotations()[dsNameTag],
			CreateTime: job.GetAnnotations()[createTimeTag],
			Status:     job.GetStatus(),
		}

		switch job.GetSource() {
		case whpb.SourceType_mysql_binlog:
			info := &whpb.MysqlBinlogJob{}
			err := info.Unmarshal(job.GetInfo())
			if err != nil {
				return nil, errors.Trace(err)
			}
			db := info.GetSource()
			ds.Type = "mysql"
			ds.PartitionNum = info.GetPartitionNum()
			ds.DB = db
		}
		dataSources = append(dataSources, ds)
	}

	return dataSources, nil
}

// SourceDB represents DB instance that need to be synced
type SourceDB struct {
	Name         string         `json:"name"`
	Type         string         `json:"type"`
	DB           whpb.DBAccount `json:"db"`
	Schemas      []*whpb.Schema `json:"sync_schema"`
	WillSyncMore bool           `json:"will_sync_more"`
	NoLocks      bool           `json:"no_locks"`
}

// TargetDB represents target database info
type TargetDB struct {
	Name string         `json:"name"`
	Type string         `json:"type"`
	DB   whpb.DBAccount `json:"db"`
}

// Pipeline represents a data sync pipeline info
type Pipeline struct {
	ID      uint64      `json:"id"`
	Name    string      `json:"name"`
	Sources []*SourceDB `json:"sources"`
	Target  *TargetDB   `json:"target"`

	Portals      []uint64          `json:"portals"`
	Stargates    []uint64          `json:"stargates"`
	PartitionNum int32             `json:"partition_nums"`
	RouteRules   []*whpb.RouteRule `json:"route_rules"`

	CreateTime string              `json:"create_time"`
	JobIDs     map[string][]uint64 `json:"job_ids"`
}

func (h *jobHandler) CreatePipeline(w http.ResponseWriter, r *http.Request) {
	pipeline := &Pipeline{}
	err := readJSON(r.Body, pipeline)
	if err != nil {
		h.r.JSON(w, http.StatusBadRequest, err.Error())
		return
	}
	log.Infof("[ether] create pipeline %+v", pipeline)

	// check pipeline
	err = validateDB(pipeline.Sources[0].DB)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] pipeline config error: %v", err))
		return
	}
	err = validateDB(pipeline.Target.DB)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] pipeline config error: %v", err))
		return
	}
	log.Infof("[ether] create pipeline %+v", pipeline)

	// create pipeline jobs
	jobs, err := h.createPipelineJobs(pipeline)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to generate jobs: %v", err))
		return
	}
	// collect job ids of this pipeline
	jobIDs := make(map[string][]uint64, len(jobs))
	for schema, sjobs := range jobs {
		for _, j := range sjobs {
			jobIDs[schema] = append(jobIDs[schema], j.GetId())
			err = h.StoreJob(j)
			if err != nil {
				h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to store jobs: %v", err))
				return
			}
		}
	}
	pipeline.JobIDs = jobIDs
	pipelineInfoBytes, err := json.Marshal(pipeline)
	if err != nil {
		h.r.JSON(w, http.StatusOK, err.Error())
		return
	}
	err = h.StorePipeline(pipeline.ID, pipelineInfoBytes)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to store pipeline %v", err))
		return
	}

	h.r.JSON(w, http.StatusOK, successResponse(pipeline))
}

func (h *jobHandler) createPipelineJobs(pipeline *Pipeline) (map[string][]*whpb.Job, error) {
	// construct labels, labels are worker ids
	var (
		portalGroupLabel   = make(map[string]*whpb.Label)
		stargateGroupLabel = make(map[string]*whpb.Label)
	)
	if len(pipeline.Portals) > 0 {
		portalGroupLabel["group"] = &whpb.Label{Vals: toGroupLabels(pipeline.Portals)}
	}
	portalGroupLabel["type"] = &whpb.Label{Vals: []string{whpb.WorkerType_Portal.String()}}
	if len(pipeline.Stargates) > 0 {
		stargateGroupLabel["group"] = &whpb.Label{Vals: toGroupLabels(pipeline.Stargates)}
	}
	stargateGroupLabel["type"] = &whpb.Label{Vals: []string{whpb.WorkerType_Stargate.String()}}

	// main job id represents the pipeline, different with datasourse id
	mainJobID, err := h.AllocID()
	if err != nil {
		return nil, errors.Trace(err)
	}
	pipeline.ID = mainJobID
	mainJobIDStr := strconv.FormatUint(mainJobID, 10)

	jobs := make(map[string][]*whpb.Job)
	source := pipeline.Sources[0]

	var (
		typeOfDump     whpb.SourceType
		typeOfBinlog   whpb.SourceType
		portalJobBytes []byte
		stargateBytes  []byte
	)
	// iterate schemas to construct jobs
	for _, schema := range source.Schemas {
		schema.NoLocks = source.NoLocks
		switch source.Type {
		case "mysql":
			typeOfDump = whpb.SourceType_mysql_dump
			typeOfBinlog = whpb.SourceType_mysql_binlog
			// portal job info
			portalJobInfo := &whpb.MysqlDumpJob{
				Source:       source.DB,
				Schemas:      []*whpb.Schema{schema},
				PartitionNum: pipeline.PartitionNum,
			}
			portalJobBytes, err = portalJobInfo.Marshal()
			if err != nil {
				return nil, errors.Trace(err)
			}
			// stargate job
			schemaInfo := whpb.SchemaInfo{
				Host:    source.DB.GetHost(),
				Port:    source.DB.GetPort(),
				Schemas: []*whpb.Schema{schema},
			}
			stargateJobInfo := &whpb.MysqlSyncJob{
				Source: schemaInfo,
				Target: pipeline.Target.DB,
				Rules:  pipeline.RouteRules,
			}
			stargateBytes, err = stargateJobInfo.Marshal()
			if err != nil {
				return nil, errors.Trace(err)
			}
		default:
			return nil, errors.Errorf("unsupported type %v", source.Type)
		}

		// whpb job for portal
		portalJobID, err := h.AllocID()
		if err != nil {
			return nil, errors.Trace(err)
		}
		portalJob := &whpb.Job{
			Id:          portalJobID,
			Type:        whpb.WorkerType_Portal,
			Source:      typeOfDump,
			Info:        portalJobBytes,
			Labels:      portalGroupLabel,
			Annotations: map[string]string{mainJobTag: mainJobIDStr, jobTypeTag: "pipeline", createTimeTag: pipeline.CreateTime},
			Status: &whpb.JobStatus{
				Id:        portalJobID,
				Type:      portalDump,
				Replicate: 1,
			},
		}

		// whpb job for dump-data stargate
		stargateDumpJobID, err := h.AllocID()
		if err != nil {
			return nil, errors.Trace(err)
		}
		replicate := computeReplicate(pipeline.PartitionNum, len(schema.Tables))
		stargateDumpJob := &whpb.Job{
			Id:          stargateDumpJobID,
			Type:        whpb.WorkerType_Stargate,
			Source:      typeOfDump,
			Info:        stargateBytes,
			Labels:      stargateGroupLabel,
			Annotations: map[string]string{mainJobTag: mainJobIDStr, jobTypeTag: "pipeline", createTimeTag: pipeline.CreateTime},
			Status: &whpb.JobStatus{
				Id:        stargateDumpJobID,
				Type:      stargateDump,
				Replicate: replicate,
			},
		}

		// whpb job for binlog data stargate
		stargateBinlogJobID, err := h.AllocID()
		if err != nil {
			return nil, errors.Trace(err)
		}
		stargateBinlogJob := &whpb.Job{
			Id:           stargateBinlogJobID,
			Type:         whpb.WorkerType_Stargate,
			Source:       typeOfBinlog,
			Info:         stargateBytes,
			Labels:       stargateGroupLabel,
			Annotations:  map[string]string{mainJobTag: mainJobIDStr, jobTypeTag: "pipeline", createTimeTag: pipeline.CreateTime},
			Dependencies: []uint64{stargateDumpJobID},
			Status: &whpb.JobStatus{
				Id:        stargateBinlogJobID,
				Type:      stargateBinlog,
				Replicate: replicate,
			},
		}

		dbName := schema.GetName()
		jobs[dbName] = append(jobs[dbName], portalJob)
		jobs[dbName] = append(jobs[dbName], stargateDumpJob)
		jobs[dbName] = append(jobs[dbName], stargateBinlogJob)
	}
	return jobs, nil
}

func (h *jobHandler) DeletePipeline(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	pipelineID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %v error %v", id, err))
		return
	}
	log.Infof("[ether] delete pipeline %d", pipelineID)

	// fetch pipeline info
	info, err := h.Handler.GetPipeline(pipelineID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to get pipeline %v, error %v", pipelineID, err))
		return
	}
	pipeline := &Pipeline{}
	err = json.Unmarshal(info, pipeline)
	if err != nil {
		log.Errorf("[ether] json unmarshal error %v, pipeline=%d data=%v", errors.ErrorStack(err), pipelineID, info)
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}
	// delete jobs belong to this pipeline
	for _, schema := range pipeline.Sources[0].Schemas {
		jobIDs := pipeline.JobIDs[schema.GetName()]
		for _, jobID := range jobIDs {
			h.DeleteJob(jobID)
		}
	}
	// delete pipeline self
	log.Infof("[ether] delete pipeline %d", pipelineID)
	err = h.Handler.DeletePipeline(pipelineID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to delete pipeline error %v", err))
		return
	}

	_, err = h.tryDeleteDatasource(&pipeline.Sources[0].DB)
	if err != nil {
		log.Errorf("try to delete datasource error %v", err)
	}
	h.r.JSON(w, http.StatusOK, successResponse(pipelineID))
}

func (h *jobHandler) GetPipelines(w http.ResponseWriter, r *http.Request) {
	pipelineInfos, err := h.Handler.GetPipelines()
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] get pipelines error: %v", err))
		return
	}

	pipelines := make([]*Pipeline, 0, len(pipelineInfos))
	for _, info := range pipelineInfos {
		pipeline := &Pipeline{}
		err := json.Unmarshal(info, pipeline)
		if err != nil {
			log.Errorf("[ether] json unmarshal error %v, data %v", errors.ErrorStack(err), string(info))
			h.r.JSON(w, http.StatusInternalServerError, err.Error())
			return
		}
		pipelines = append(pipelines, pipeline)
	}

	h.r.JSON(w, http.StatusOK, successResponse(pipelines))
}

func (h *jobHandler) GetPipeline(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	pipelineID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %v error %v", id, err))
		return
	}

	// fetch pipeline info
	pipelineInfo, err := h.Handler.GetPipeline(pipelineID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to get job %v, error %v", id, err))
		return
	}
	if pipelineInfo == nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] job %v not found", id))
		return
	}
	pipeline := &Pipeline{}
	err = json.Unmarshal(pipelineInfo, pipeline)
	if err != nil {
		log.Errorf("[ether] json unmarshal error %v, data %v", errors.ErrorStack(err), pipelineInfo)
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}
	h.r.JSON(w, http.StatusOK, successResponse(pipeline))
}

// StageSummary represents stage summary
type StageSummary struct {
	Percent uint64 `json:"percent"`
	QPS     uint64 `json:"qps"`
}

// PipelineSummary represents pipeline summary status
type PipelineSummary struct {
	IsInitialized         bool         `json:"is_initialized"`
	PortalDumpSummary     StageSummary `json:"portal_dump_summary"`
	PortalBinlogSummary   StageSummary `json:"portal_binlog_summary"`
	StargateDumpSummary   StageSummary `json:"stargate_dump_summary"`
	StargateBinlogSummary StageSummary `json:"stargate_binlog_summary"`
}

func (h *jobHandler) GetPipelineSummary(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	pipelineID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %v error", id, err))
		return
	}

	// fetch pipeline
	pipelineInfo, err := h.Handler.GetPipeline(pipelineID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to get job %v, error %v", id, err))
		return
	}
	if pipelineInfo == nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] job %v not found", id))
		return
	}
	pipeline := &Pipeline{}
	err = json.Unmarshal(pipelineInfo, pipeline)
	if err != nil {
		log.Errorf("[ether] json unmarshal error %v, data %v", errors.ErrorStack(err), pipelineInfo)
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}

	// construct pipeline summary status
	var status PipelineSummary
	// fetch binlog job
	binlogJob, err := h.FindSameDataSourceJob(&pipeline.Sources[0].DB)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to fetch portal binlog job: %v", err))
		return
	}
	if binlogJob == nil {
		h.r.JSON(w, http.StatusOK, notFoundResponsef("[ether] not found datasource"))
		return
	}

	status.IsInitialized = binlogJob.GetStatus().GetStage() == whpb.JobStage_Running
	status.PortalBinlogSummary.QPS = binlogJob.GetStatus().GetQps()
	// fetch pipeline jobs
	stageJobs, err := h.GetJobsByStage(pipeline)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] get job by stage error %v", err))
		return
	}
	for stage, jobs := range stageJobs {
		var (
			total    uint64
			progress uint64
			qps      uint64
		)
		for _, j := range jobs {
			js := j.GetStatus()
			total += js.GetTotal()
			progress += js.GetTotal() * js.GetPercent()
			qps += js.GetQps()
		}

		var summary *StageSummary
		switch stage {
		case portalDump:
			summary = &status.PortalDumpSummary
		case stargateDump:
			summary = &status.StargateDumpSummary
		case stargateBinlog:
			summary = &status.StargateBinlogSummary
		}

		if total != 0 {
			summary.Percent = progress / total
		} else {
			summary.Percent = 0
		}
		summary.QPS = qps
		log.Debugf("[ether] pipeline stage %v summary %+v", stage, summary)
	}
	log.Debugf("pipeline status %+v", status)
	h.r.JSON(w, http.StatusOK, successResponse(status))
}

// fetchJobsByStage fetch jobs according to stage
func (h *jobHandler) GetJobsByStage(p *Pipeline) (map[string][]*whpb.Job, error) {
	var (
		portalDumpJobs     []*whpb.Job
		stargateDumpJobs   []*whpb.Job
		stargateBinlogJobs []*whpb.Job
	)
	for _, sjobs := range p.JobIDs {
		jobs, err := h.GetJobs(sjobs)
		if err != nil {
			return nil, errors.Trace(err)
		}
		for _, j := range jobs {
			switch j.Type {
			case whpb.WorkerType_Portal:
				portalDumpJobs = append(portalDumpJobs, j)
			case whpb.WorkerType_Stargate:
				switch j.Source {
				case whpb.SourceType_mysql_dump:
					stargateDumpJobs = append(stargateDumpJobs, j)
				case whpb.SourceType_mysql_binlog:
					stargateBinlogJobs = append(stargateBinlogJobs, j)
				}
			}
		}
	}

	stageJobs := make(map[string][]*whpb.Job)
	stageJobs[portalDump] = portalDumpJobs
	stageJobs[stargateDump] = stargateDumpJobs
	stageJobs[stargateBinlog] = stargateBinlogJobs
	return stageJobs, nil
}

// PipelineDetail represents pipeline detail status
type PipelineDetail struct {
	SourceStatus *whpb.JobStatus              `json:"source_status"`
	SchemaStatus map[string][]*whpb.JobStatus `json:"schema_status"`
}

func (h *jobHandler) GetPipelineDetail(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	pipelineID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %v error", id, err))
		return
	}

	// fetch pipeline info
	pipelineInfo, err := h.Handler.GetPipeline(pipelineID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to get job %v, error %v", id, err))
		return
	}
	if pipelineInfo == nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] job %v not found", id))
		return
	}
	pipeline := &Pipeline{}
	err = json.Unmarshal(pipelineInfo, pipeline)
	if err != nil {
		log.Errorf("[ether] json unmarshal error %v, data %v", errors.ErrorStack(err), pipelineInfo)
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}

	// fetch binlog job
	binlogJob, err := h.FindSameDataSourceJob(&pipeline.Sources[0].DB)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to fetch portal binlog job: %v", err))
		return
	}
	if binlogJob == nil {
		h.r.JSON(w, http.StatusOK, notFoundResponsef("[ether] not found datasource"))
		return
	}

	// fetch pipeline jobs
	var jobs []*whpb.Job
	schemaline := make(map[string][]*whpb.Job)
	for schema, jobIDs := range pipeline.JobIDs {
		jobs, err = h.GetJobs(jobIDs)
		if err != nil {
			h.r.JSON(w, http.StatusOK, errResponsef("[ether] get jobs error %v", err))
			return
		}
		schemaline[schema] = jobs
	}

	// construct pipeline detail status
	var status PipelineDetail
	js, err := formatJobStatus([]*whpb.Job{binlogJob})
	if err != nil {
		log.Errorf("[ether] format mysql job error %v, job %+v", errors.ErrorStack(err), binlogJob)
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}
	status.SourceStatus = js[0]
	status.SchemaStatus = make(map[string][]*whpb.JobStatus)
	for schema, jobs := range schemaline {
		jss, err := formatJobStatus(jobs)
		if err != nil {
			log.Errorf("[ether] format mysql job error %v, job %+v", errors.ErrorStack(err), jobs)
			h.r.JSON(w, http.StatusInternalServerError, err.Error())
			return
		}
		status.SchemaStatus[schema] = jss
	}
	h.r.JSON(w, http.StatusOK, successResponse(status))
}

func (h *jobHandler) GetJobStatus(w http.ResponseWriter, r *http.Request) {
	// parse job ids
	idsStr := mux.Vars(r)["ids"]
	idStrs := strings.Split(idsStr, ",")
	ids := make([]uint64, len(idStrs))
	for index, idStr := range idStrs {
		jobID, err := strconv.ParseUint(idStr, 10, 64)
		if err != nil {
			h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job ids %s error %v", jobID, err))
			return
		}
		ids[index] = jobID
	}

	// fetch job status
	jobs, err := h.GetJobs(ids)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] get job status error %v", err))
		return
	}

	jss, err := formatJobStatus(jobs)
	if err != nil {
		log.Errorf("[ether] format mysql job error %v, job %+v", errors.ErrorStack(err), jobs)
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}
	h.r.JSON(w, http.StatusOK, successResponse(jss))
}

func (h *jobHandler) Replicate(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	jobID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %s error %v", id, err))
		return
	}

	replicateStr := mux.Vars(r)["replicate"]
	replicate, err := strconv.ParseInt(replicateStr, 10, 32)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse replicate %s error %v", replicate, err))
		return
	}

	log.Infof("[ether] modify job %d replicate to %d", jobID, replicate)
	err = h.AdjustReplicate(jobID, int32(replicate))
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to adjust replicate error %v", err))
		return
	}

	h.r.JSON(w, http.StatusOK, successResponse(jobID))
}

func (h *jobHandler) Action(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	jobID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %s error %v", id, err))
		return
	}
	action := mux.Vars(r)["action"]

	log.Infof("[ether] apply action %s on %d", action, jobID)

	switch action {
	case "pause":
		err = h.PauseJob(jobID)
	case "restart":
		err = h.ReStartJob(jobID)
	default:
		err = errors.Errorf("[ether] unsupport action %s", action)
	}

	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %s error %v", id, err))
	} else {
		h.r.JSON(w, http.StatusOK, successResponse(jobID))
	}
}

func (h *jobHandler) Delete(w http.ResponseWriter, r *http.Request) {
	id := mux.Vars(r)["id"]
	jobID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse job id %s error %v", id, err))
		return
	}

	log.Infof("[ether] delete job %d", jobID)
	err = h.DeleteJob(jobID)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to delete job error %v", err))
		return
	}
	h.r.JSON(w, http.StatusOK, successResponse(jobID))
}

func (h *jobHandler) SchemaInfo(w http.ResponseWriter, r *http.Request) {
	host := mux.Vars(r)["host"]
	portStr := mux.Vars(r)["port"]
	port, err := strconv.ParseInt(portStr, 10, 32)
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] parse host %s error %v", host, err))
		return
	}

	schemaInfo, err := h.GetSchemaInfo(host, int32(port))
	if err != nil {
		h.r.JSON(w, http.StatusOK, errResponsef("[ether] fail to get schemaInfo %v", err))
		return
	}
	if schemas := schemaInfo.GetSchemas(); schemas != nil {
		h.r.JSON(w, http.StatusOK, successResponse(schemas))
	} else {
		h.r.JSON(w, http.StatusOK, successResponse([]*whpb.Schema{}))
	}
}

func (h *jobHandler) GetResource(w http.ResponseWriter, r *http.Request) {
	resourceType := mux.Vars(r)["type"]
	id := mux.Vars(r)["id"]
	resourceID, err := strconv.ParseUint(id, 10, 64)
	if err != nil {
		h.r.JSON(w, http.StatusBadRequest, err.Error())
		return
	}

	raw, err := h.Resource(resourceType, resourceID)
	if err != nil {
		log.Errorf("get resource error %v, type %v, resouceID %v", errors.ErrorStack(err), resourceType, resourceID)
		h.r.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}

	h.r.JSON(w, http.StatusOK, raw)
}

func (h *jobHandler) tryDeleteDatasource(source *whpb.DBAccount) (bool, error) {
	pipelineInfos, err := h.Handler.GetPipelines()
	if err != nil {
		log.Errorf("[ether] get pipelines error %v", errors.ErrorStack(err))
		return false, errors.Annotate(err, "get pipelines")
	}

	for _, info := range pipelineInfos {
		pipeline := &Pipeline{}
		err = json.Unmarshal(info, pipeline)
		if err != nil {
			log.Errorf("[ether] json unmarshal error %v, data %v", errors.ErrorStack(err), info)
			return false, errors.Annotate(err, "json unmarshal")
		}

		if len(pipeline.Sources) == 0 {
			log.Errorf("[ether] pipeline %s source is empty", info)
			return false, errors.Errorf("pipeline %d source is empty", pipeline.ID)
		}

		// same datasources have same host and port
		t := &pipeline.Sources[0].DB
		if t.GetHost() == source.GetHost() && t.GetPort() == source.GetPort() {
			return false, nil
		}
	}

	job, err := h.Handler.FindSameDataSourceJob(source)
	if err != nil {
		log.Errorf("[ether] find same datasource %+v error %v", source, err)
		return false, errors.Annotatef(err, "find same datasource %+v error %v", source)
	}
	if job == nil {
		log.Errorf("[ether] not found datasorce %v", source)
		return false, errors.NotFoundf("[ether] not found datasource")
	}

	err = h.DeleteJob(job.Id)
	if err != nil {
		log.Errorf("[ether] fail to delete datasource error %v", err)
		return false, errors.Annotatef(err, "fail to delete datasource", job.Id)
	}

	return true, nil
}