package ether

import (
	"io"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/whpb"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
)

// notLeaderError is returned when current server is not the leader and not possible to process request.
// TODO: work as proxy.
var notLeaderError = grpc.Errorf(codes.Unavailable, "not leader")
var waitFenceInterval = 1 * time.Second

const (
	maxHeartbeatSyncInterval = 100
)

// WorkerHeartbeat implements whpb server WorkerHeartbeat interface.
func (s *Server) WorkerHeartbeat(server whpb.Ether_WorkerHeartbeatServer) error {
	cluster := s.GetCluster()
	if cluster == nil {
		header := s.errorResponseHeader(0, whpb.ErrorCode_NotLeader, "cluster is not boostrapped")
		err := server.Send(&whpb.WorkerHeartbeatResponse{Header: header})
		return errors.Trace(err)
	}

	isNew := true
	for {
		request, err := server.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return errors.Trace(err)
		}

		if err = s.validateRequest(request.GetHeader(), true, cluster.cached); err != nil {
			return errors.Trace(err)
		}

		workerID := request.GetHeader().GetWorkerId()
		hbStreams := cluster.controller.hbStreams
		if isNew {
			hbStreams.bindStream(workerID, server)
			isNew = false
		}

		log.Debugf("[ether] worker heart worker %d", workerID)

		// renew worker lease
		jobStatus := request.GetJobs()
		err = cluster.cached.handleWorkerHeartbeat(workerID, jobStatus)
		if err != nil {
			log.Errorf("[ether] refresh worker error %v", err)
		}
	}
}

// RegisterWorker registers worker
func (s *Server) RegisterWorker(ctx context.Context, in *whpb.RegisterWorkerRequest) (*whpb.RegisterWorkerResponse, error) {
	cluster := s.GetCluster()
	if cluster == nil {
		log.Error("[ether] cluster is not boorstrapped")
		header := s.errorResponseHeader(0, whpb.ErrorCode_NotLeader, "cluster is not boorstrapped")
		return &whpb.RegisterWorkerResponse{Header: header}, nil
	}

	if err := s.validateRequest(in.GetHeader(), false, nil); err != nil {
		return nil, errors.Trace(err)
	}

	worker := in.GetWorker()
	err := cluster.cached.createWorker(worker)
	if err != nil {
		log.Errorf("[ether] alloc id error %v", err)
		header := s.errorResponseHeader(0, whpb.ErrorCode_CreateNewWorker, err.Error())
		return &whpb.RegisterWorkerResponse{Header: header}, nil
	}

	return &whpb.RegisterWorkerResponse{Header: &whpb.ResponseHeader{ClusterId: s.clusterID, WorkerId: worker.GetId()}}, nil
}

// RenewJob renews job.
func (s *Server) RenewJob(ctx context.Context, in *whpb.RenewJobRequest) (*whpb.RenewJobResponse, error) {
	cluster := s.GetCluster()
	if cluster == nil {
		log.Error("[ether] cluster is not boorstrapped")
		header := s.errorResponseHeader(0, whpb.ErrorCode_NotLeader, "cluster is not boorstrapped")
		return &whpb.RenewJobResponse{Header: header}, nil
	}

	if err := s.validateRequest(in.GetHeader(), true, cluster.cached); err != nil {
		return nil, errors.Trace(err)
	}

	workerID := in.GetHeader().GetWorkerId()
	jobIDs := in.GetJobIds()
	log.Debugf("[ether] worker %d renew jobs %+v", workerID, jobIDs)

	var (
		expiredJobs []uint64
		errJobs     []uint64
	)

	// check contract
	// in normal case we wouldn't meet it:
	// before contract expired, worker would give up the contract
	for _, jobID := range jobIDs {
		// renew job lease
		jobLease := in.GetLease()
		if jobLease < contractLease {
			jobLease = contractLease
		}
		// fixme: renew in a batch
		isExpired, err := cluster.cached.renewContract(jobID, workerID, extendLease(jobLease))
		if err != nil {
			errJobs = append(errJobs, jobID)
		}
		if isExpired {
			expiredJobs = append(expiredJobs, jobID)
		}
	}

	return &whpb.RenewJobResponse{Header: &whpb.ResponseHeader{ClusterId: s.clusterID, WorkerId: workerID}, ErrorJobs: errJobs, ExpiredJobs: expiredJobs}, nil
}

// CreateFence implements grpc interface.
func (s *Server) CreateFence(ctx context.Context, in *whpb.CreateFenceRequest) (*whpb.CreateFenceResponse, error) {
	cluster := s.GetCluster()
	if cluster == nil {
		header := s.errorResponseHeader(0, whpb.ErrorCode_NotLeader, "cluster is not boorstrapped")
		return &whpb.CreateFenceResponse{Header: header}, nil
	}

	if err := s.validateRequest(in.GetHeader(), false, nil); err != nil {
		return nil, errors.Trace(err)
	}

	// get topic name
	topics := in.GetTopics()
	if topics == nil {
		return nil, grpc.Errorf(codes.InvalidArgument, "topic is empty")
	}
	log.Infof("[ether] create fence for topics %+v", topics)

	fence, err := cluster.cached.createFence(topics)
	if err != nil {
		return nil, errors.Trace(err)
	}

	// query partitions to save to etcd
	response := &whpb.CreateFenceResponse{
		Header: &whpb.ResponseHeader{
			ClusterId: s.clusterID,
		},
		FenceId: fence.Id,
	}
	return response, nil
}

// WaitOnFenceMidway implements grpc interface.
func (s *Server) WaitOnFenceMidway(in *whpb.WaitOnFenceRequest, srv whpb.Ether_WaitOnFenceMidwayServer) error {
	return s.WaitOnFence(in, srv, false)
}

// WaitOnFenceFinal implements grpc interface.
func (s *Server) WaitOnFenceFinal(in *whpb.WaitOnFenceRequest, srv whpb.Ether_WaitOnFenceFinalServer) error {
	return s.WaitOnFence(in, srv, true)
}

// GetClusterInfo implements grpc interface.
func (s *Server) GetClusterInfo(ctx context.Context, in *whpb.GetClusterInfoRequest) (*whpb.GetClusterInfoResponse, error) {
	leader, err := s.GetLeader()
	if err != nil {
		return nil, grpc.Errorf(codes.Unknown, err.Error())
	}

	members, err := GetMembers(s.client)
	if err != nil {
		return nil, grpc.Errorf(codes.Unknown, err.Error())
	}

	return &whpb.GetClusterInfoResponse{
		ClusterId:       s.clusterID,
		StorageRootPath: s.getClusterRootPath(),
		Leader:          leader,
		Members:         members,
	}, nil
}

// WaitOnFenceServer waits fence on server
type WaitOnFenceServer interface {
	Send(*whpb.WaitOnFenceResponse) error
}

// WaitOnFence waits fence
func (s *Server) WaitOnFence(in *whpb.WaitOnFenceRequest, srv WaitOnFenceServer, isFinal bool) error {
	cluster := s.GetCluster()
	if cluster == nil {
		header := s.errorResponseHeader(0, whpb.ErrorCode_NotLeader, "cluster is not boorstrapped")
		err := srv.Send(&whpb.WaitOnFenceResponse{Header: header})
		return errors.Trace(err)
	}

	if err := s.validateRequest(in.GetHeader(), false, nil); err != nil {
		return errors.Trace(err)
	}

	fenceID := in.GetFenceId()
	if fenceID == 0 {
		return grpc.Errorf(codes.InvalidArgument, "fenceId is empty")
	}
	jobID := in.GetJobId()
	if jobID == 0 {
		return grpc.Errorf(codes.InvalidArgument, "job_id is empty")
	}

	topic := in.GetTopic()
	if topic == nil {
		return grpc.Errorf(codes.InvalidArgument, "topic is empty")
	}

	// get partitions
	partitions := in.GetPartitions()
	pids := make([]uint64, len(partitions))
	for i, p := range partitions {
		pids[i] = uint64(p)
	}

	err := cluster.cached.completePartitions(fenceID, jobID, topic, pids, isFinal)
	if err != nil {
		return errors.Trace(err)
	}

	for {
		midway, final := cluster.cached.isFenceSynced(fenceID, jobID)
		if (isFinal && final) || (!isFinal && midway) {
			err = srv.Send(&whpb.WaitOnFenceResponse{Header: &whpb.ResponseHeader{ClusterId: s.clusterID}, IsSynced: true})
			return errors.Trace(err)
		}

		time.Sleep(waitFenceInterval)
	}
}

func (s *Server) errorResponseHeader(workerID uint64, errCode whpb.ErrorCode, errMsg string) *whpb.ResponseHeader {
	reponseErr := &whpb.Error{
		Errcode: errCode,
		Message: errMsg,
	}
	return &whpb.ResponseHeader{
		ClusterId: s.clusterID,
		WorkerId:  workerID,
		Error:     reponseErr,
	}
}

// validateRequest checks if Server is leader and clusterID is matched.
// TODO: Call it in gRPC intercepter.
func (s *Server) validateRequest(header *whpb.RequestHeader, validateWorker bool, cluster *clusterInfo) error {
	if !s.IsLeader() {
		return notLeaderError
	}
	if header.GetClusterId() != s.clusterID {
		return grpc.Errorf(codes.FailedPrecondition, "mismatch cluster id, need %d but got %d", s.clusterID, header.GetClusterId())
	}

	if validateWorker {
		workerID := header.GetWorkerId()
		worker := cluster.getWorker(workerID)
		if workerID == 0 || worker == nil {
			return errors.Errorf("workerid %d not found in cluster %d", workerID, header.GetClusterId())
		}
	}

	return nil
}