package ether

import (
	"fmt"

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

var errClusterNotBootstrap = errors.New("not found cluster")

// it's a tag represents job type that created by front
// like `datasource`, `pipeline`
const jobTypeTag = "jobTypeForFront"

// Handler is a helper to export methods to handle API requests.
type Handler struct {
	s *Server
}

func newHandler(s *Server) *Handler {
	return &Handler{s: s}
}

// AllocID returns new allocted id
func (h *Handler) AllocID() (uint64, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return 0, errClusterNotBootstrap
	}

	id, err := cluster.cached.id.Alloc()
	return id, errors.Trace(err)
}

// GetWorkers returns all workers
func (h *Handler) GetWorkers() ([]*whpb.Worker, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	workersInfo := cluster.cached.allWorkers()
	workers := make([]*whpb.Worker, 0, len(workersInfo))
	for _, worker := range workersInfo {
		workers = append(workers, worker.getWorker())
	}

	return workers, nil
}

// GetWorker returns the specified worker
func (h *Handler) GetWorker(workerID uint64) (*whpb.Worker, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	workerInfo := cluster.cached.getWorker(workerID)
	if workerInfo == nil {
		return nil, errors.NotFoundf("worker %d", workerID)
	}

	return workerInfo.Worker, nil
}

// GetDataSourceJobs returns jobs that associated datasource
func (h *Handler) GetDataSourceJobs() ([]*whpb.Job, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	jobInfos := cluster.cached.allJobs()
	if len(jobInfos) == 0 {
		return nil, nil
	}

	jobs := make([]*whpb.Job, 0, len(jobInfos))
	for _, job := range jobInfos {
		//job must have job.GetAnnotations() in logic
		if job.GetAnnotations()[jobTypeTag] == "dataSource" {
			jobs = append(jobs, job.getJob())
		}
	}

	return jobs, nil
}

// GetJob returns specified job in ether
func (h *Handler) GetJob(jobID uint64) (*whpb.Job, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	job := cluster.cached.getJob(jobID)
	if job == nil {
		return nil, errors.NotFoundf("[ether] not found job %d", jobID)
	}

	return job.getJob(), nil
}

// FindSameDataSourceJob returns same data source job
func (h *Handler) FindSameDataSourceJob(db *whpb.DBAccount) (*whpb.Job, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	job := cluster.cached.findPortalJob(db.GetHost(), db.GetPort())
	if job == nil {
		return nil, nil
	}
	return job.getJob(), nil
}

// StoreJob stores job
func (h *Handler) StoreJob(job *whpb.Job) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}

	err := cluster.cached.storeJob(job)
	return errors.Trace(err)
}

// StorePipeline store pipeline info
func (h *Handler) StorePipeline(jobID uint64, info []byte) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}

	err := cluster.cached.storePipeline(jobID, info)
	return errors.Trace(err)
}

// GetPipelines returns all origin pipelines
func (h *Handler) GetPipelines() (map[uint64][]byte, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	pipelines := cluster.cached.allPipelines()
	return pipelines, nil
}

// GetPipeline returns origin pipeline
func (h *Handler) GetPipeline(id uint64) ([]byte, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	info := cluster.cached.getPipeline(id)
	if info == nil {
		return info, errors.NotFoundf("pipeline %d", id)
	}

	return info, nil
}

// DeletePipeline return delete op result
func (h *Handler) DeletePipeline(id uint64) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}
	err := cluster.cached.deletePipeline(id)
	return errors.Trace(err)
}

// AdjustReplicate adjusts job's replicate
func (h *Handler) AdjustReplicate(jobID uint64, replicate int32) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}

	return cluster.cached.adjustReplicate(jobID, replicate)
}

// GetJobs returns Jobs
func (h *Handler) GetJobs(jobIDs []uint64) ([]*whpb.Job, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	jobs := make([]*whpb.Job, 0, len(jobIDs))
	for _, jobID := range jobIDs {
		job := cluster.cached.getJob(jobID)
		if job != nil {
			jobs = append(jobs, job.getJob())
		}
	}

	return jobs, nil
}

// PauseJob pauses job
func (h *Handler) PauseJob(jobID uint64) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}

	err := cluster.cached.pauseJob(jobID)
	return errors.Trace(err)
}

// ReStartJob starts job
func (h *Handler) ReStartJob(jobID uint64) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}

	err := cluster.cached.restartJob(jobID)
	return errors.Trace(err)
}

// DeleteJob deletes job
func (h *Handler) DeleteJob(jobID uint64) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}

	log.Infof("[ether] delete job %d from cache", jobID)
	err := cluster.cached.deleteJob(jobID)
	return errors.Trace(err)
}

// GetSchemaInfo returns schemaInfo
func (h *Handler) GetSchemaInfo(host string, port int32) (*whpb.SchemaInfo, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return nil, errClusterNotBootstrap
	}

	schemaInfo := &whpb.SchemaInfo{}
	ok, err := cluster.cached.kv.loadSchemaInfo(host, port, schemaInfo)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if !ok {
		return nil, errors.Errorf("[ether] DB host %s port %d is not initialized", host, port)
	}

	return schemaInfo, nil
}

// DeleteSchemaInfo return op result
func (h *Handler) DeleteSchemaInfo(host string, port int32) error {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return errClusterNotBootstrap
	}

	err := cluster.cached.kv.deleteSchemaInfo(host, port)
	return errors.Trace(err)
}

// Resource returns raw resource
func (h *Handler) Resource(tp string, id uint64) (string, error) {
	cluster := h.s.GetCluster()
	if cluster == nil {
		return "", errClusterNotBootstrap
	}

	switch tp {
	case "job":
		job := cluster.cached.getJob(id)
		return fmt.Sprintf("%+v", job), nil
	case "worker":
		worker := cluster.cached.getWorker(id)
		return fmt.Sprintf("%+v", worker), nil
	default:
		return "unsupported type", nil
	}
}