// Ack acknowledges a task in the queue.
func (c *Client) Ack(ctx context.Context, queue string, task server.TaskID) (
	err error) {
	_, call := c.GoAck(queue, task, nil)
	select {
	case res := <-call:
		return res.Error
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (n *MultiNode) waitElection(ctx context.Context, group uint64) {
	ch := make(chan struct{})
	n.pmu.Lock()
	n.pendingElects[group] = append(n.pendingElects[group], ch)
	n.pmu.Unlock()
	select {
	case <-ch:
	case <-ctx.Done():
	case <-n.done:
	}
}

// DeQ dequeues a task from the given queue. It returns the task or an error.
func (c *Client) DeQ(ctx context.Context, queue string) (task server.Task,
	err error) {

	_, call := c.GoDeQ(queue, nil)
	select {
	case res := <-call:
		if res.Error != nil {
			return server.Task{}, res.Error
		}
		return res.Data.(server.Task), nil
	case <-ctx.Done():
		return server.Task{}, ctx.Err()
	}
}

// EnQ enqueues a task with the given body in the queue. It returns the
// task's id or an error.
func (c *Client) EnQ(ctx context.Context, queue string, body []byte) (
	id server.TaskID, err error) {

	_, call := c.GoEnQ(queue, body, nil)
	select {
	case res := <-call:
		if res.Error != nil {
			return 0, res.Error
		}
		return res.Data.(server.TaskID), res.Error
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

// StepBatch steps all messages in the batch. Context is used for enqueuing the
// batch and timeout is used as the maximum wait time when stepping messages
// in the batch.
func (n *MultiNode) StepBatch(ctx context.Context, batch Batch,
	timeout time.Duration) error {

	bt := batchTimeout{
		batch:   batch,
		timeout: timeout,
	}
	select {
	case n.recvc <- bt:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (mn *multiNode) step(ctx context.Context, m multiMessage) error {
	ch := mn.recvc
	if m.msg.Type == pb.MsgProp {
		ch = mn.propc
	}

	select {
	case ch <- m:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-mn.done:
		return ErrStopped
	}
}

// Step advances the state machine using msgs. The ctx.Err() will be returned,
// if any.
func (n *node) step(ctx context.Context, m pb.Message) error {
	ch := n.recvc
	if m.Type == pb.MsgProp {
		ch = n.propc
	}

	select {
	case ch <- m:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
}

func (n *MultiNode) Exists(ctx context.Context, gid uint64) (ok bool) {
	ch := make(chan groupResponse, 1)
	n.groupc <- groupRequest{
		reqType: groupRequestStatus,
		group:   &group{id: gid},
		ch:      ch,
	}
	select {
	case res := <-ch:
		return res.err == nil
	case <-ctx.Done():
		return false
	case <-n.done:
		return false
	}
}

// Sync processes a synchrounous request and returns the response and error.
func (h *hive) Sync(ctx context.Context, req interface{}) (res interface{},
	err error) {

	id := uint64(rand.Int63())
	ch := make(chan syncRes, 1)

	// We should run this in parallel in case we are blocked on h.syncCh.
	go func() {
		sc := syncReqAndChan{
			req: syncReq{ID: id, Data: req},
			ch:  ch,
		}
		select {
		case h.syncCh <- sc:
		case <-ctx.Done():
		}
	}()

	select {
	case r := <-ch:
		if r.Err != nil {
			return nil, errors.New(r.Err.Error())
		}
		return r.Data, nil

	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func (n *MultiNode) processConfChange(ctx context.Context, group uint64,
	cc raftpb.ConfChange, gn GroupNode) error {

	if group == 0 || gn.Node == 0 || gn.Group != group {
		glog.Fatalf("invalid group node: %v", gn)
	}

	id := n.genID()
	req := Request{Data: gn}

	var err error
	cc.Context, err = n.encReq(id, req)
	if err != nil {
		return err
	}

	ch := n.line.wait(id, req)

	d, err := cc.Marshal()
	if err != nil {
		return err
	}
	select {
	case n.propc <- multiMessage{
		group: group,
		msg: raftpb.Message{
			Type:    raftpb.MsgProp,
			Entries: []raftpb.Entry{{Type: raftpb.EntryConfChange, Data: d}},
		},
	}:
	case <-ctx.Done():
		n.line.cancel(id)
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}

	select {
	case res := <-ch:
		return res.Err
	case <-ctx.Done():
		n.line.cancel(id)
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
}

// Propose proposes the request and returns the response. This method blocks and
// returns either when the ctx is cancelled or the raft node returns a response.
func (n *MultiNode) Propose(ctx context.Context, group uint64,
	req interface{}) (res interface{}, err error) {

	id := n.genID()
	r := Request{
		Data: req,
	}

	d, err := n.encReq(id, r)
	if err != nil {
		return nil, err
	}

	glog.V(2).Infof("%v waits on raft request %v", n, id)
	ch := n.line.wait(id, r)
	mm := multiMessage{group,
		raftpb.Message{
			Type:    raftpb.MsgProp,
			Entries: []raftpb.Entry{{Data: d}},
		}}
	select {
	case n.propc <- mm:
	case <-ctx.Done():
		n.line.cancel(id)
		return nil, ctx.Err()
	case <-n.done:
		return nil, ErrStopped
	}

	select {
	case res := <-ch:
		glog.V(2).Infof("%v wakes up for raft request %v", n, id)
		return res.Data, res.Err
	case <-ctx.Done():
		n.line.cancel(id)
		return nil, ctx.Err()
	case <-n.done:
		return nil, ErrStopped
	}
}

func (n *MultiNode) CreateGroup(ctx context.Context, cfg GroupConfig) error {
	glog.V(2).Infof("creating a new group %v (%v) on node %v (%v) with peers %v",
		cfg.ID, cfg.Name, n.id, n.name, cfg.Peers)

	rs, ds, _, lei, _, err := OpenStorage(cfg.ID, cfg.DataDir, cfg.StateMachine)
	if err != nil {
		glog.Fatalf("cannot open storage: %v", err)
	}

	snap, err := rs.Snapshot()
	if err != nil {
		glog.Errorf("error in storage snapshot: %v", err)
		return err
	}

	n.gen.StartFrom((lei + cfg.SnapCount) << 8) // To avoid conflicts.

	c := &etcdraft.Config{
		ID:              cfg.ID,
		ElectionTick:    cfg.ElectionTicks,
		HeartbeatTick:   cfg.HeartbeatTicks,
		Storage:         rs,
		MaxSizePerMsg:   cfg.MaxMsgSize,
		MaxInflightMsgs: cfg.MaxInFlights,
		// TODO(soheil): Figure this one out:
		//               Applied: lsi,
	}
	g := &group{
		node:         n,
		id:           cfg.ID,
		name:         cfg.Name,
		stateMachine: cfg.StateMachine,
		raftStorage:  rs,
		diskStorage:  ds,
		applyc:       make(chan etcdraft.Ready, cfg.SnapCount),
		savec:        make(chan readySaved, 1),
		fsyncTime:    cfg.FsyncTick,
		snapCount:    cfg.SnapCount,
		snapped:      snap.Metadata.Index,
		applied:      snap.Metadata.Index,
		confState:    snap.Metadata.ConfState,
		stopc:        make(chan struct{}),
		applierDone:  make(chan struct{}),
		saverDone:    make(chan struct{}),
	}
	ch := make(chan groupResponse, 1)
	n.groupc <- groupRequest{
		reqType: groupRequestCreate,
		group:   g,
		config:  c,
		peers:   cfg.Peers,
		ch:      ch,
	}
	select {
	case res := <-ch:
		return res.err
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
}