package ether

import (
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"

	"golang.org/x/net/context"
)

const (
	schedulerInterval = 3 * time.Second
)

// Scheduler is an interface to schedule jobs.
type Scheduler interface {
	Name() string
	Schedule(cluster *clusterInfo) Command
}

// jobScheduler assigns jobs to suitable workers.
type jobScheduler struct {
	name      string
	algorithm ScheduleAlgorithm
}

func newJobScheduler(name string) *jobScheduler {
	return &jobScheduler{
		name:      name,
		algorithm: new(simpleFirstFit),
	}
}

func (s *jobScheduler) Name() string {
	return s.name
}

func (s *jobScheduler) Schedule(cluster *clusterInfo) Command {
	var jobs Resources
	var workers Resources

	// query all jobs
	jobInfos := cluster.allJobs()
	for _, job := range jobInfos {
		if cluster.canSchedule(job.GetId()) {
			jobs = append(jobs, job)
		}
	}

	ws := cluster.allWorkers()
	for _, worker := range ws {
		if worker.isOnline() {
			workers = append(workers, worker)
		}
	}

	resultSet := s.algorithm.Select(jobs, workers)
	for _, res := range resultSet {
		jobID := res[0].GetId()
		workerID := res[1].GetId()
		cmd, err := s.createNewContract(cluster, jobID, workerID)
		if err == nil {
			return cmd
		}
		log.Errorf("[ether] create new contract error %v", err)
	}
	return nil
}

func (s *jobScheduler) createNewContract(cluster *clusterInfo, jobID, workerID uint64) (Command, error) {
	err := cluster.createContract(jobID, workerID, extendLease(contractLease))
	if err != nil {
		return nil, errors.Trace(err)
	}
	worker := cluster.getWorker(workerID)
	if worker == nil {
		return nil, nil
	}
	job := cluster.getJob(jobID)
	if job == nil {
		return nil, nil
	}
	return newSignContractCmd(worker, job), nil
}

// rebalanceScheduler grab jobs from workers to rebalance distribution of jobs.
type rebalanceScheduler struct {
	name      string
	algorithm ScheduleAlgorithm
}

func newRebalanceScheduler(name string) *rebalanceScheduler {
	return &rebalanceScheduler{
		name:      name,
		algorithm: new(simpleExpel),
	}
}

func (s *rebalanceScheduler) Name() string {
	return s.name
}

func (s *rebalanceScheduler) Schedule(cluster *clusterInfo) Command {
	var jobs Resources
	var workers Resources

	// query all jobs
	jobInfos := cluster.allJobs()
	for _, job := range jobInfos {
		if job.canRunning() {
			continue
		}
		jobs = append(jobs, job)
	}

	ws := cluster.allWorkers()
	for _, worker := range ws {
		if worker.isOnline() {
			workers = append(workers, worker)
		}
	}

	resultSet := s.algorithm.Select(jobs, workers)
	for _, res := range resultSet {
		jobID := res[0].GetId()
		workerID := res[1].GetId()
		cmd, err := s.releaseContract(cluster, jobID, workerID)
		if err == nil {
			return cmd
		}
		log.Errorf("[ether] create new contract error %v", err)
	}
	return nil
}

func (s *rebalanceScheduler) releaseContract(cluster *clusterInfo, jobID, workerID uint64) (Command, error) {
	cluster.terminateContract(jobID, workerID)
	worker := cluster.getWorker(workerID)
	if worker != nil {
		return nil, nil
	}
	job := cluster.getJob(jobID)
	if job != nil {
		return nil, nil
	}
	return newReleaseContract(worker, job), nil
}

// scheduleController controls scheduler.
type scheduleController struct {
	Scheduler
	nextInterval time.Duration
	minInterval  time.Duration
	ctx          context.Context
	cancel       context.CancelFunc
}

func newScheduleController(ctx context.Context, s Scheduler, minInterval time.Duration) *scheduleController {
	ctx, cancel := context.WithCancel(ctx)
	return &scheduleController{
		Scheduler:    s,
		nextInterval: minInterval,
		minInterval:  minInterval,
		ctx:          ctx,
		cancel:       cancel,
	}
}

func (s *scheduleController) Ctx() context.Context {
	return s.ctx
}

func (s *scheduleController) Stop() {
	s.cancel()
}

func (s *scheduleController) Schedule(cluster *clusterInfo) Command {
	s.nextInterval = s.minInterval
	cmd := s.Scheduler.Schedule(cluster)
	if cmd != nil {
		return cmd
	}

	// if we have no schedule, increase the interval exponentially.
	s.nextInterval = minDuration(time.Duration(float64(s.nextInterval)*scheduleIntervalFactor), maxScheduleInterval)
	return nil
}

func (s *scheduleController) GetInterval() time.Duration {
	return s.nextInterval
}