package ether

import (
	"context"
	"sync"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	"github.com/pingcap/wormhole/pkg/whpb"
)

const (
	regionheartbeatSendChanCap = 1024
	maxScheduleRetries         = 10
	scheduleIntervalFactor     = 1.3
	maxScheduleInterval        = time.Minute
	minScheduleInterval        = time.Second
)

type controller struct {
	sync.RWMutex

	wg     sync.WaitGroup
	ctx    context.Context
	cancel context.CancelFunc

	client   *clientv3.Client
	cluster  *clusterInfo
	rootPath string

	pause          bool
	schedulers     map[string]*scheduleController
	backgroundJobs map[string]BackgroundJob

	hbStreams *heartbeatStreams
}

func newController(cluster *clusterInfo, hbStreams *heartbeatStreams, client *clientv3.Client, rootPath string) *controller {
	ctx, cancel := context.WithCancel(context.Background())
	return &controller{
		ctx:            ctx,
		cancel:         cancel,
		client:         client,
		cluster:        cluster,
		rootPath:       rootPath,
		schedulers:     make(map[string]*scheduleController),
		backgroundJobs: make(map[string]BackgroundJob),
		hbStreams:      hbStreams,
	}
}

func (c *controller) run() {
	c.addBackgroundJob(newBackgoundWatchJob(c.ctx, c.client, "watch_expired_contract", etcdutil.ContractPathPrefix(c.rootPath), c.handleExpiredContractEvent))
	if err := c.cluster.loadContracts(); err != nil {
		log.Errorf("[ether] load contracts error %v", err)
		c.stop()
		return
	}
	c.addScheduler(newJobScheduler("job_scheduler"), minScheduleInterval)
	c.addScheduler(newRebalanceScheduler("rebalance_job_scheduler"), 15*minScheduleInterval)
}

func (c *controller) stop() {
	c.cancel()
	c.wg.Wait()
}

func (c *controller) getSchedulers() []string {
	c.RLock()
	names := make([]string, len(c.schedulers))
	index := 0
	for name := range c.schedulers {
		names[index] = name
		index++
	}
	c.RUnlock()
	return names
}

func (c *controller) addScheduler(scheduler Scheduler, interval time.Duration) error {
	c.Lock()
	if _, ok := c.schedulers[scheduler.Name()]; ok {
		c.Unlock()
		return errors.Errorf("[ether] scheduler %v exists", scheduler.Name())
	}

	s := newScheduleController(c.ctx, scheduler, interval)
	go c.runScheduler(s)
	c.schedulers[scheduler.Name()] = s
	c.Unlock()
	return nil
}

func (c *controller) removeScheduler(name string) error {
	c.Lock()
	s, ok := c.schedulers[name]
	if !ok {
		c.Unlock()
		return errors.Errorf("[ether] scheduler %v not found", name)
	}
	delete(c.schedulers, name)
	c.Unlock()

	s.Stop()
	return nil
}

func (c *controller) runScheduler(s *scheduleController) {
	c.wg.Add(1)
	defer c.wg.Done()

	timer := time.NewTimer(s.GetInterval())
	defer timer.Stop()
	for {
		select {
		case <-timer.C:
			timer.Reset(s.GetInterval())
			if cmd := s.Schedule(c.cluster); cmd != nil {
				c.dispatchCommand(cmd)
			}

		case <-s.Ctx().Done():
			log.Infof("[ether] name of scheduleController: %v stopped: %v", s.Name(), s.Ctx().Err())
			return
		}
	}
}

func (c *controller) dispatchCommand(cmd Command) {
	response := cmd.Generate()
	c.hbStreams.sendMsg(response)
}

func (c *controller) getBackgroundJobs() []string {
	c.RLock()
	names := make([]string, len(c.backgroundJobs))
	index := 0
	for name := range c.backgroundJobs {
		names[index] = name
		index++
	}
	c.RUnlock()
	return names
}

func (c *controller) addBackgroundJob(job BackgroundJob) error {
	c.Lock()
	if _, ok := c.backgroundJobs[job.Name()]; ok {
		c.Unlock()
		return errors.Errorf("[ether] background job %v exists", job.Name())
	}

	go c.runBackgroundJob(job)
	c.backgroundJobs[job.Name()] = job
	c.Unlock()
	return nil
}

func (c *controller) removeBackgroundJob(name string) error {
	c.Lock()
	job, ok := c.backgroundJobs[name]
	if !ok {
		c.Unlock()
		return errors.Errorf("[ether] scheduler %v not found", name)
	}

	job.Stop()
	delete(c.backgroundJobs, name)
	c.Unlock()
	return nil
}

func (c *controller) runBackgroundJob(job BackgroundJob) {
	c.wg.Add(1)
	defer c.wg.Done()

	job.Run()
}

func (c *controller) handleExpiredContractEvent(event *clientv3.Event) error {
	jobID, workerID, err := parseContractKey(event.Kv)
	if err != nil {
		return errors.Trace(err)
	}

	if event.Type == mvccpb.PUT {
		return nil
	}

	log.Infof("[ether] revoke contract job %d vs worker %d", jobID, workerID)
	// update cluster cache
	c.cluster.terminateContract(jobID, workerID)
	return nil
}

// grpc stream
type heartBeatStream interface {
	Send(*whpb.WorkerHeartbeatResponse) error
}

type streamUpdate struct {
	workerID uint64
	stream   heartBeatStream
}

// heartbeatStreams manage worker's hb connections
// send message from here
type heartbeatStreams struct {
	wg     sync.WaitGroup
	ctx    context.Context
	cancel context.CancelFunc

	clusterID      uint64
	streams        map[uint64]heartBeatStream
	msgCh          chan *whpb.WorkerHeartbeatResponse
	streamUpdateCh chan streamUpdate
}

func newHeartbeatStreams(clusterID uint64) *heartbeatStreams {
	ctx, cancel := context.WithCancel(context.Background())
	hs := &heartbeatStreams{
		ctx:            ctx,
		cancel:         cancel,
		clusterID:      clusterID,
		streams:        make(map[uint64]heartBeatStream),
		msgCh:          make(chan *whpb.WorkerHeartbeatResponse, regionheartbeatSendChanCap),
		streamUpdateCh: make(chan streamUpdate),
	}
	go hs.run()
	return hs
}

func (s *heartbeatStreams) Close() {
	s.cancel()
	s.wg.Wait()
}

func (s *heartbeatStreams) run() {
	s.wg.Add(1)
	defer s.wg.Done()
	for {
		select {
		case stream := <-s.streamUpdateCh:
			s.streams[stream.workerID] = stream.stream
		case msg := <-s.msgCh:
			workerID := msg.GetHeader().GetWorkerId()
			if stream, ok := s.streams[workerID]; ok {
				if err := stream.Send(msg); err != nil {
					log.Errorf("[worker %v] send heart response message error: %v", workerID, err)
				}
			} else {
				log.Warningf("[worker %v] not found stream to send message, skip message", workerID)
			}
		case <-s.ctx.Done():
			return
		}
	}
}

func (s *heartbeatStreams) bindStream(workerID uint64, stream heartBeatStream) {
	su := streamUpdate{
		workerID: workerID,
		stream:   stream,
	}
	select {
	case s.streamUpdateCh <- su:
	case <-s.ctx.Done():
	}
}

func (s *heartbeatStreams) sendMsg(msg *whpb.WorkerHeartbeatResponse) {
	msg.Header.ClusterId = s.clusterID

	select {
	case s.msgCh <- msg:
	case <-s.ctx.Done():
	}
}

func (s *heartbeatStreams) sendErr(workerID uint64, errCode whpb.ErrorCode, errMsg string) {
	msg := &whpb.WorkerHeartbeatResponse{
		Header: &whpb.ResponseHeader{
			ClusterId: s.clusterID,
			WorkerId:  workerID,
			Error: &whpb.Error{
				Errcode: errCode,
				Message: errMsg,
			},
		},
	}

	select {
	case s.msgCh <- msg:
	case <-s.ctx.Done():
	}
}