func (n *node) processSnapshot(s raftpb.Snapshot) {
	lead := n.raft.Status().Lead
	if lead == 0 {
		return
	}
	addr := n.peers.Get(lead)
	x.AssertTruef(addr != "", "Should have the leader address: %v", lead)
	pool := pools().get(addr)
	x.AssertTruef(pool != nil, "Leader: %d pool should not be nil", lead)

	_, err := populateShard(context.TODO(), pool, 0)
	x.Checkf(err, "processSnapshot")
}

// returns true if the geometry represented by uid/attr intersects the given loop or point
func (q QueryData) intersects(g types.Geo) bool {
	x.AssertTruef(q.pt != nil || q.loop != nil, "At least a point or loop should be defined.")
	switch v := g.T.(type) {
	case *geom.Point:
		p := pointFromPoint(v)
		if q.pt != nil {
			// Points only intersect if they are the same. (We allow for small rounding errors)
			return q.pt.ApproxEqual(p)
		}
		// else loop is not nil
		return q.loop.ContainsPoint(p)

	case *geom.Polygon:
		l, err := loopFromPolygon(v)
		if err != nil {
			return false
		}
		if q.pt != nil {
			return l.ContainsPoint(*q.pt)
		}
		// else loop is not nil
		return Intersects(l, q.loop)
	default:
		// A type that we don't know how to handle.
		return false
	}
}

// FilterUids filters the uids based on the corresponding values and QueryData.
func FilterUids(uids *task.List, values []*task.Value, q *QueryData) *task.List {
	x.AssertTruef(len(values) == len(uids.Uids), "lengths not matching")
	rv := &task.List{}
	for i := 0; i < len(values); i++ {
		valBytes := values[i].Val
		if bytes.Equal(valBytes, nil) {
			continue
		}
		vType := values[i].ValType
		if types.TypeID(vType) != types.GeoID {
			continue
		}
		var g types.Geo
		if err := g.UnmarshalBinary(valBytes); err != nil {
			continue
		}

		if !q.MatchesFilter(g) {
			continue
		}

		// we matched the geo filter, add the uid to the list
		rv.Uids = append(rv.Uids, uids.Uids[i])
	}
	return rv
}

func (p *proposals) Store(pid uint32, ch chan error) {
	p.Lock()
	defer p.Unlock()
	_, has := p.ids[pid]
	x.AssertTruef(!has, "Same proposal is being stored again.")
	p.ids[pid] = ch
}

// ServeTask is used to respond to a query.
func (w *grpcWorker) ServeTask(ctx context.Context, q *task.Query) (*task.Result, error) {
	if ctx.Err() != nil {
		return &emptyResult, ctx.Err()
	}

	gid := group.BelongsTo(q.Attr)
	x.Trace(ctx, "Attribute: %q NumUids: %v groupId: %v ServeTask", q.Attr, len(q.Uids), gid)

	var reply *task.Result
	x.AssertTruef(groups().ServesGroup(gid),
		"attr: %q groupId: %v Request sent to wrong server.", q.Attr, gid)

	c := make(chan error, 1)
	go func() {
		var err error
		reply, err = processTask(q)
		c <- err
	}()

	select {
	case <-ctx.Done():
		return reply, ctx.Err()
	case err := <-c:
		return reply, err
	}
}

// Sort is used to sort given UID matrix.
func (w *grpcWorker) Sort(ctx context.Context, s *task.Sort) (*task.SortResult, error) {
	if ctx.Err() != nil {
		return &emptySortResult, ctx.Err()
	}

	gid := group.BelongsTo(s.Attr)
	//x.Trace(ctx, "Attribute: %q NumUids: %v groupId: %v Sort", q.Attr(), q.UidsLength(), gid)

	var reply *task.SortResult
	x.AssertTruef(groups().ServesGroup(gid),
		"attr: %q groupId: %v Request sent to wrong server.", s.Attr, gid)

	c := make(chan error, 1)
	go func() {
		var err error
		reply, err = processSort(s)
		c <- err
	}()

	select {
	case <-ctx.Done():
		return &emptySortResult, ctx.Err()
	case err := <-c:
		return reply, err
	}
}

func addIndexMutation(ctx context.Context, attr, token string,
	tokensTable *TokensTable, edge *task.DirectedEdge, del bool) {
	key := x.IndexKey(attr, token)
	plist, decr := GetOrCreate(key)
	defer decr()

	x.AssertTruef(plist != nil, "plist is nil [%s] %d %s",
		token, edge.ValueId, edge.Attr)
	if del {
		_, err := plist.AddMutation(ctx, edge, Del)
		if err != nil {
			x.TraceError(ctx, x.Wrapf(err,
				"Error deleting %s for attr %s entity %d: %v",
				token, edge.Attr, edge.Entity))
		}
		indexLog.Printf("DEL [%s] [%d] OldTerm [%s]",
			edge.Attr, edge.Entity, token)

	} else {
		_, err := plist.AddMutation(ctx, edge, Set)
		if err != nil {
			x.TraceError(ctx, x.Wrapf(err,
				"Error adding %s for attr %s entity %d: %v",
				token, edge.Attr, edge.Entity))
		}
		indexLog.Printf("SET [%s] [%d] NewTerm [%s]",
			edge.Attr, edge.Entity, token)

		tokensTable.Add(token)
	}
}

func (l *List) decr() {
	val := atomic.AddInt32(&l.refcount, -1)
	x.AssertTruef(val >= 0, "List reference should never be less than zero: %v", val)
	if val > 0 {
		return
	}
	listPool.Put(l)
}

func getTokens(funcArgs []string) ([]string, error) {
	x.AssertTruef(len(funcArgs) > 1, "Invalid function")
	if len(funcArgs) != 2 {
		return nil, x.Errorf("Function requires 2 arguments, but got %d",
			len(funcArgs))
	}
	return getStringTokens(funcArgs[1])
}

func parsePeer(peer string) (uint64, string) {
	x.AssertTrue(len(peer) > 0)

	kv := strings.SplitN(peer, ":", 2)
	x.AssertTruef(len(kv) == 2, "Invalid peer format: %v", peer)
	pid, err := strconv.ParseUint(kv[0], 10, 64)
	x.Checkf(err, "Invalid peer id: %v", kv[0])
	// TODO: Validate the url kv[1]
	return pid, kv[1]
}

func parentCoverTokens(parents s2.CellUnion, cover s2.CellUnion) []string {
	// We index parents and cover using different prefix, that makes it more
	// performant at query time to only look up parents/cover depending on what
	// kind of query it is.
	tokens := make([]string, 0, len(parents)+len(cover))
	tokens = appendTokens(tokens, parents, parentPrefix)
	tokens = appendTokens(tokens, cover, coverPrefix)
	x.AssertTruef(len(tokens) == len(parents)+len(cover), "%d %d %d",
		len(tokens), len(parents), len(cover))
	return tokens
}

func newPosting(t *task.DirectedEdge, op uint32) *types.Posting {
	x.AssertTruef(bytes.Equal(t.Value, nil) || t.ValueId == math.MaxUint64,
		"This should have been set by the caller.")

	return &types.Posting{
		Uid:     t.ValueId,
		Value:   t.Value,
		ValType: uint32(t.ValueType),
		Label:   t.Label,
		Op:      op,
	}
}

func (n *node) send(m raftpb.Message) {
	x.AssertTruef(n.id != m.To, "Seding message to itself")
	data, err := m.Marshal()
	x.Check(err)
	if m.Type != raftpb.MsgHeartbeat && m.Type != raftpb.MsgHeartbeatResp {
		fmt.Printf("\t\tSENDING: %v %v-->%v\n", m.Type, m.From, m.To)
	}
	select {
	case n.messages <- sendmsg{to: m.To, data: data}:
		// pass
	default:
		log.Fatalf("Unable to push messages to channel in send")
	}
}

func (g *groupi) newNode(groupId uint32, nodeId uint64, publicAddr string) *node {
	g.Lock()
	defer g.Unlock()
	if g.local == nil {
		g.local = make(map[uint32]*node)
	}

	node := newNode(groupId, nodeId, publicAddr)
	if _, has := g.local[groupId]; has {
		x.AssertTruef(false, "Didn't expect a node in RAFT group mapping: %v", groupId)
	}
	g.local[groupId] = node
	return node
}

// processSort does either a coarse or a fine sort.
func processSort(ts *task.Sort) (*task.SortResult, error) {
	attr := ts.Attr
	x.AssertTruef(ts.Count > 0,
		("We do not yet support negative or infinite count with sorting: %s %d. " +
			"Try flipping order and return first few elements instead."),
		attr, ts.Count)

	n := len(ts.UidMatrix)
	out := make([]intersectedList, n)
	for i := 0; i < n; i++ {
		// offsets[i] is the offset for i-th posting list. It gets decremented as we
		// iterate over buckets.
		out[i].offset = int(ts.Offset)
		out[i].ulist = &task.List{Uids: []uint64{}}
	}

	// Iterate over every bucket in TokensTable.
	t := posting.GetTokensTable(attr)

	var token string
	if ts.Desc {
		token = t.GetLast()
	} else {
		token = t.GetFirst()
	}
BUCKETS:
	for len(token) > 0 {
		err := intersectBucket(ts, attr, token, out)
		switch err {
		case errDone:
			break BUCKETS
		case errContinue:
			// Continue iterating over tokens.
		default:
			return &emptySortResult, err
		}
		if ts.Desc {
			token = t.GetPrev(token)
		} else {
			token = t.GetNext(token)
		}
	}

	r := new(task.SortResult)
	for _, il := range out {
		r.UidMatrix = append(r.UidMatrix, il.ulist)
	}
	return r, nil
}

func (n *node) AddToCluster(ctx context.Context, pid uint64) error {
	addr := n.peers.Get(pid)
	x.AssertTruef(len(addr) > 0, "Unable to find conn pool for peer: %d", pid)
	rc := &task.RaftContext{
		Addr:  addr,
		Group: n.raftContext.Group,
		Id:    pid,
	}
	rcBytes, err := rc.Marshal()
	x.Check(err)
	return n.raft.ProposeConfChange(ctx, raftpb.ConfChange{
		ID:      pid,
		Type:    raftpb.ConfChangeAddNode,
		NodeID:  pid,
		Context: rcBytes,
	})
}

// returns true if the geometry represented by g is within the given loop or cap
func (q QueryData) isWithin(g types.Geo) bool {
	x.AssertTruef(q.pt != nil || q.loop != nil || q.cap != nil, "At least a point, loop or cap should be defined.")
	gpt, ok := g.T.(*geom.Point)
	if !ok {
		// We will only consider points for within queries.
		return false
	}

	s2pt := pointFromPoint(gpt)
	if q.pt != nil {
		return q.pt.ApproxEqual(s2pt)
	}

	if q.loop != nil {
		return q.loop.ContainsPoint(s2pt)
	}
	return q.cap.ContainsPoint(s2pt)
}

// AssignUidsOverNetwork assigns new uids and writes them to the umap.
func AssignUidsOverNetwork(ctx context.Context, umap map[string]uint64) error {
	gid := group.BelongsTo("_uid_")
	num := createNumQuery(gid, umap)

	var ul *task.List
	var err error
	if groups().ServesGroup(gid) {
		ul, err = assignUids(ctx, num)
		if err != nil {
			return err
		}

	} else {
		_, addr := groups().Leader(gid)
		p := pools().get(addr)
		conn, err := p.Get()
		if err != nil {
			x.TraceError(ctx, x.Wrapf(err, "Error while retrieving connection"))
			return err
		}
		defer p.Put(conn)

		c := NewWorkerClient(conn)
		ul, err = c.AssignUids(ctx, num)
		if err != nil {
			x.TraceError(ctx, x.Wrapf(err, "Error while getting uids"))
			return err
		}
	}

	x.AssertTruef(len(ul.Uids) == int(num.Val),
		"Requested: %d != Retrieved Uids: %d", num.Val, len(ul.Uids))

	i := 0
	for k, v := range umap {
		if v == 0 {
			uid := ul.Uids[i]
			umap[k] = uid // Write uids to map.
			i++
		}
	}
	return nil
}

func parsePredicates(groupId uint32, p string) error {
	preds := strings.Split(p, ",")
	x.AssertTruef(len(preds) > 0, "Length of predicates in config should be > 0")

	for _, pred := range preds {
		pred = strings.TrimSpace(pred)
		meta := predMeta{
			val: pred,
			gid: groupId,
		}
		if strings.HasSuffix(pred, "*") {
			meta.val = strings.TrimSuffix(meta.val, "*")
		} else {
			meta.exactMatch = true
		}
		groupConfig.pred = append(groupConfig.pred, meta)
	}
	return nil
}

// returns true if the geometry represented by uid/attr contains the given point
func (q QueryData) contains(g types.Geo) bool {
	x.AssertTruef(q.pt != nil || q.loop != nil, "At least a point or loop should be defined.")
	if q.loop != nil {
		// We don't support polygons containing polygons yet.
		return false
	}

	poly, ok := g.T.(*geom.Polygon)
	if !ok {
		// We will only consider polygons for contains queries.
		return false
	}

	s2loop, err := loopFromPolygon(poly)
	if err != nil {
		return false
	}
	return s2loop.ContainsPoint(*q.pt)
}