// parseMutationOp parses and stores set or delete operation string in Mutation.
func parseMutationOp(l *lex.Lexer, op string, mu *Mutation) error {
	if mu == nil {
		return x.Errorf("Mutation is nil.")
	}

	parse := false
	for item := range l.Items {
		if item.Typ == itemText {
			continue
		}
		if item.Typ == itemLeftCurl {
			if parse {
				return x.Errorf("Too many left curls in set mutation.")
			}
			parse = true
		}
		if item.Typ == itemMutationContent {
			if !parse {
				return x.Errorf("Mutation syntax invalid.")
			}
			if op == "set" {
				mu.Set = item.Val
			} else if op == "delete" {
				mu.Del = item.Val
			} else {
				return x.Errorf("Invalid mutation operation.")
			}
		}
		if item.Typ == itemRightCurl {
			return nil
		}
	}
	return x.Errorf("Invalid mutation formatting.")
}

// parseArguments parses the arguments part of the GraphQL query root.
func parseArguments(l *lex.Lexer) (result []pair, rerr error) {
	for {
		var p pair

		// Get key.
		item := <-l.Items
		if item.Typ == itemArgName {
			p.Key = item.Val

		} else if item.Typ == itemRightRound {
			break

		} else {
			return result, x.Errorf("Expecting argument name. Got: %v", item)
		}

		// Get value.
		item = <-l.Items
		if item.Typ != itemArgVal {
			return result, x.Errorf("Expecting argument value. Got: %v", item)
		}

		p.Val = item.Val
		result = append(result, p)
	}
	return result, nil
}

// getFragment parses a fragment definition (not reference).
func getFragment(l *lex.Lexer) (*fragmentNode, error) {
	var name string
	for item := range l.Items {
		if item.Typ == itemText {
			v := strings.TrimSpace(item.Val)
			if len(v) > 0 && name == "" {
				// Currently, we take the first nontrivial token as the
				// fragment name and ignore everything after that until we see
				// a left curl.
				name = v
			}
		} else if item.Typ == itemLeftCurl {
			break
		} else {
			return nil, x.Errorf("Unexpected item in fragment: %v %v", item.Typ, item.Val)
		}
	}
	if name == "" {
		return nil, x.Errorf("Empty fragment name")
	}

	gq := &GraphQuery{
		Args: make(map[string]string),
	}
	if err := godeep(l, gq); err != nil {
		return nil, err
	}
	fn := &fragmentNode{
		Name: name,
		Gq:   gq,
	}
	return fn, nil
}

func parseFunction(l *lex.Lexer) (*Function, error) {
	var g *Function
	for item := range l.Items {
		if item.Typ == itemFilterFunc { // Value.
			g = &Function{Name: item.Val}
			itemInFunc := <-l.Items
			if itemInFunc.Typ != itemLeftRound {
				return nil, x.Errorf("Expected ( after func name [%s]", g.Name)
			}
			for itemInFunc = range l.Items {
				if itemInFunc.Typ == itemRightRound {
					break
				} else if itemInFunc.Typ != itemFilterFuncArg {
					return nil, x.Errorf("Expected arg after func [%s], but got item %v",
						g.Name, itemInFunc)
				}
				if len(g.Attr) == 0 {
					g.Attr = itemInFunc.Val
				} else {
					g.Args = append(g.Args, itemInFunc.Val)
				}
			}
		} else if item.Typ == itemRightRound {
			break
		} else {
			return nil, x.Errorf("Expected a function but got %q", item.Val)
		}
	}
	return g, nil
}

// Parse initializes and runs the lexer. It also constructs the GraphQuery subgraph
// from the lexed items.
func Parse(input string) (gq *GraphQuery, mu *Mutation, rerr error) {
	l := &lex.Lexer{}
	query, vmap, err := parseQueryWithVariables(input)
	if err != nil {
		return nil, nil, err
	}

	l.Init(query)
	go run(l)

	fmap := make(fragmentMap)
	for item := range l.Items {
		switch item.Typ {
		case lex.ItemError:
			return nil, nil, x.Errorf(item.Val)
		case itemText:
			continue

		case itemOpType:
			if item.Val == "mutation" {
				if mu != nil {
					return nil, nil, x.Errorf("Only one mutation block allowed.")
				}
				if mu, rerr = getMutation(l); rerr != nil {
					return nil, nil, rerr
				}
			} else if item.Val == "fragment" {
				// TODO(jchiu0): This is to be done in ParseSchema once it is ready.
				fnode, rerr := getFragment(l)
				if rerr != nil {
					return nil, nil, rerr
				}
				fmap[fnode.Name] = fnode
			} else if item.Val == "query" {
				if gq, rerr = getVariablesAndQuery(l, vmap); rerr != nil {
					return nil, nil, rerr
				}
			}
		case itemLeftCurl:
			if gq, rerr = getQuery(l); rerr != nil {
				return nil, nil, rerr
			}
		}
	}

	if gq != nil {
		// Try expanding fragments using fragment map.
		if err := gq.expandFragments(fmap); err != nil {
			return nil, nil, err
		}

		// Substitute all variables with corresponding values
		if err := substituteVariables(gq, vmap); err != nil {
			return nil, nil, err
		}
	}

	return gq, mu, nil
}

// PopulateShard gets data for predicate pred from server with id serverId and
// writes it to RocksDB.
func populateShard(ctx context.Context, pl *pool, group uint32) (int, error) {
	gkeys, err := generateGroup(group)
	if err != nil {
		return 0, x.Wrapf(err, "While generating keys group")
	}

	conn, err := pl.Get()
	if err != nil {
		return 0, err
	}
	defer pl.Put(conn)
	c := NewWorkerClient(conn)

	stream, err := c.PredicateData(context.Background(), gkeys)
	if err != nil {
		return 0, err
	}
	x.Trace(ctx, "Streaming data for group: %v", group)

	kvs := make(chan *task.KV, 1000)
	che := make(chan error)
	go writeBatch(ctx, kvs, che)

	// We can use count to check the number of posting lists returned in tests.
	count := 0
	for {
		kv, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			close(kvs)
			return count, err
		}
		count++

		// We check for errors, if there are no errors we send value to channel.
		select {
		case kvs <- kv:
			// OK
		case <-ctx.Done():
			x.TraceError(ctx, x.Errorf("Context timed out while streaming group: %v", group))
			close(kvs)
			return count, ctx.Err()
		case err := <-che:
			x.TraceError(ctx, x.Errorf("Error while doing a batch write for group: %v", group))
			close(kvs)
			return count, err
		}
	}
	close(kvs)

	if err := <-che; err != nil {
		x.TraceError(ctx, x.Errorf("Error while doing a batch write for group: %v", group))
		return count, err
	}
	x.Trace(ctx, "Streaming complete for group: %v", group)
	return count, nil
}

func main() {
	flag.Parse()
	logrus.SetLevel(logrus.DebugLevel)
	var srcl, dstl []R
	f, bufReader := getReader(*src)
	var err error

	srcCount := 0
	var strBuf bytes.Buffer
	for {
		err = x.ReadLine(bufReader, &strBuf)
		if err != nil {
			break
		}
		srcCount++
		rnq, err := rdf.Parse(strBuf.String())
		x.Checkf(err, "Unable to parse line: [%v]", strBuf.String())
		srcl = append(srcl, convert(rnq))
	}
	if err != nil && err != io.EOF {
		err := x.Errorf("Error while reading file: %v", err)
		log.Fatalf("%+v", err)
	}
	x.Check(f.Close())
	fmt.Println("Source done")

	f, bufReader = getReader(*dst)
	dstCount := 0
	for {
		err = x.ReadLine(bufReader, &strBuf)
		if err != nil {
			break
		}
		dstCount++
		rnq, err := rdf.Parse(strBuf.String())
		x.Checkf(err, "Unable to parse line: [%v]", strBuf.String())
		dstl = append(dstl, convert(rnq))
	}
	if err != nil && err != io.EOF {
		err := x.Errorf("Error while reading file: %v", err)
		log.Fatalf("%+v", err)
	}
	x.Check(f.Close())

	fmt.Printf("Src: [%d] Dst: [%d]\n", srcCount, dstCount)
	sort.Sort(ByR(srcl))
	sort.Sort(ByR(dstl))
	fmt.Println("Comparing now")
	//for i := 0; i < 100; i++ {
	//fmt.Printf("[S,D] %v %v\n", srcl[i], dstl[i])
	//}
	compare(srcl, dstl)
}

// getRoot gets the root graph query object after parsing the args.
func getRoot(l *lex.Lexer) (gq *GraphQuery, rerr error) {
	gq = &GraphQuery{
		Args: make(map[string]string),
	}
	item := <-l.Items
	if item.Typ != itemName {
		return nil, x.Errorf("Expected some name. Got: %v", item)
	}

	gq.Alias = item.Val
	item = <-l.Items
	if item.Typ != itemLeftRound {
		return nil, x.Errorf("Expected variable start. Got: %v", item)
	}

	item = <-l.Items
	if item.Typ == itemGenerator {
		// Store the generator function.
		gen, err := parseFunction(l)
		if !schema.IsIndexed(gen.Attr) {
			return nil, x.Errorf(
				"Field %s is not indexed and cannot be used in functions",
				gen.Attr)
		}
		if err != nil {
			return nil, err
		}
		gq.Func = gen
	} else if item.Typ == itemArgument {
		args, err := parseArguments(l)
		if err != nil {
			return nil, err
		}
		for _, p := range args {
			if p.Key == "_uid_" {
				gq.UID, rerr = strconv.ParseUint(p.Val, 0, 64)
				if rerr != nil {
					return nil, rerr
				}
			} else if p.Key == "_xid_" {
				gq.XID = p.Val
			} else {
				return nil, x.Errorf("Expecting _uid_ or _xid_. Got: %+v", p)
			}
		}
	} else {
		return nil, x.Errorf("Unexpected root argument.")
	}

	return gq, nil
}

func (gq *GraphQuery) expandFragments(fmap fragmentMap) error {
	// We have to make a copy of children to preserve order and replace
	// fragment references with fragment content. The copy is newChildren.
	var newChildren []*GraphQuery
	// Expand non-fragments. Do not append to gq.Children.
	for _, child := range gq.Children {
		if child.isFragment() {
			fname := child.fragment // Name of fragment being referenced.
			fchild := fmap[fname]
			if fchild == nil {
				return x.Errorf("Missing fragment: %s", fname)
			}
			if err := fchild.expand(fmap); err != nil {
				return err
			}
			newChildren = append(newChildren, fchild.Gq.Children...)
		} else {
			if err := child.expandFragments(fmap); err != nil {
				return err
			}
			newChildren = append(newChildren, child)
		}
	}
	gq.Children = newChildren
	return nil
}

// Parse parses the schema file.
func Parse(file string) (rerr error) {
	b, err := ioutil.ReadFile(file)
	if err != nil {
		return x.Errorf("Error reading file: %v", err)
	}
	return ParseBytes(b)
}

// IndexTokens return tokens, without the predicate prefix and index rune.
func IndexTokens(attr string, p types.Value) ([]string, error) {
	schemaType := schema.TypeOf(attr)
	if !schemaType.IsScalar() {
		return nil, x.Errorf("Cannot index attribute %s of type object.", attr)
	}
	s := schemaType.(types.Scalar)
	schemaVal, err := s.Convert(p)
	if err != nil {
		return nil, err
	}
	switch v := schemaVal.(type) {
	case *types.Geo:
		return geo.IndexTokens(v)
	case *types.Int32:
		return types.IntIndex(attr, v)
	case *types.Float:
		return types.FloatIndex(attr, v)
	case *types.Date:
		return types.DateIndex(attr, v)
	case *types.Time:
		return types.TimeIndex(attr, v)
	case *types.String:
		return types.DefaultIndexKeys(attr, v), nil
	}
	return nil, nil
}

func BackupOverNetwork(ctx context.Context) error {
	// If we haven't even had a single membership update, don't run backup.
	if len(*peer) > 0 && groups().LastUpdate() == 0 {
		x.Trace(ctx, "This server hasn't yet been fully initiated. Please retry later.")
		return x.Errorf("Uninitiated server. Please retry later")
	}
	// Let's first collect all groups.
	gids := groups().KnownGroups()

	ch := make(chan *BackupPayload, len(gids))
	for _, gid := range gids {
		go func(group uint32) {
			reqId := uint64(rand.Int63())
			ch <- handleBackupForGroup(ctx, reqId, group)
		}(gid)
	}

	for i := 0; i < len(gids); i++ {
		bp := <-ch
		if bp.Status != BackupPayload_SUCCESS {
			x.Trace(ctx, "Backup status: %v for group id: %d", bp.Status, bp.GroupId)
			return fmt.Errorf("Backup status: %v for group id: %d", bp.Status, bp.GroupId)
		} else {
			x.Trace(ctx, "Backup successful for group: %v", bp.GroupId)
		}
	}
	x.Trace(ctx, "DONE backup")
	return nil
}

func (v *Int32) fromFloat(f float64) error {
	if f > math.MaxInt32 || f < math.MinInt32 || math.IsNaN(f) {
		return x.Errorf("Float out of int32 range")
	}
	*v = Int32(f)
	return nil
}

func (w *grpcWorker) RaftMessage(ctx context.Context, query *Payload) (*Payload, error) {
	if ctx.Err() != nil {
		return &Payload{}, ctx.Err()
	}

	for idx := 0; idx < len(query.Data); {
		sz := int(binary.LittleEndian.Uint32(query.Data[idx : idx+4]))
		idx += 4
		msg := raftpb.Message{}
		if idx+sz-1 > len(query.Data) {
			return &Payload{}, x.Errorf(
				"Invalid query. Size specified: %v. Size of array: %v\n", sz, len(query.Data))
		}
		if err := msg.Unmarshal(query.Data[idx : idx+sz]); err != nil {
			x.Check(err)
		}
		if msg.Type != raftpb.MsgHeartbeat && msg.Type != raftpb.MsgHeartbeatResp {
			fmt.Printf("RECEIVED: %v %v-->%v\n", msg.Type, msg.From, msg.To)
		}
		if err := w.applyMessage(ctx, msg); err != nil {
			return &Payload{}, err
		}
		idx += sz
	}
	// fmt.Printf("Got %d messages\n", count)
	return &Payload{}, nil
}

// Sort sorts the given array in-place.
func (s Scalar) Sort(v []Value, ul *task.List, desc bool) error {
	b := sortBase{v, ul}
	var toBeSorted sort.Interface
	switch s.ID() {
	case DateID:
		toBeSorted = byDate{b}
	case DateTimeID:
		toBeSorted = byDateTime{b}
	case Int32ID:
		toBeSorted = byInt32{b}
	case FloatID:
		toBeSorted = byFloat{b}
	case StringID:
		toBeSorted = byString{b}
	case BytesID:
		toBeSorted = byByteArray{b}
	default:
		return x.Errorf("Scalar doesn't support sorting %s", s)
	}
	if desc {
		toBeSorted = sort.Reverse(toBeSorted)
	}
	sort.Sort(toBeSorted)
	return nil
}

// UnmarshalBinary unmarshals the data from a binary format.
func (v *Int32) UnmarshalBinary(data []byte) error {
	if len(data) < 4 {
		return x.Errorf("Invalid data for int32 %v", data)
	}
	*v = Int32(binary.LittleEndian.Uint32(data))
	return nil
}

func processScalar(l *lex.Lexer) error {
	for item := range l.Items {
		switch item.Typ {
		case itemLeftRound:
			{
				return processScalarBlock(l)
			}
		case itemScalarName:
			{
				var name, typ string
				name = item.Val

				next := <-l.Items
				if next.Typ != itemCollon {
					return x.Errorf("Missing collon")
				}

				next = <-l.Items
				if next.Typ != itemScalarType {
					return x.Errorf("Missing Type")
				}
				typ = next.Val

				if t, ok := getScalar(typ); ok {
					str[name] = t
				} else {
					return x.Errorf("Invalid type")
				}

				// Check for index.
				next = <-l.Items
				if next.Typ == itemAt {
					index := <-l.Items
					if index.Typ == itemIndex {
						indexedFields[name] = true
					} else {
						return x.Errorf("Invalid index specification")
					}
				}
				return nil
			}
		case lex.ItemError:
			return x.Errorf(item.Val)
		}
	}
	return nil
}

// UnmarshalBinary unmarshals the data from a binary format.
func (v *Date) UnmarshalBinary(data []byte) error {
	if len(data) < 8 {
		return x.Errorf("Invalid data for date %v", data)
	}
	val := binary.LittleEndian.Uint64(data)
	tm := time.Unix(int64(val), 0)
	return v.fromTime(tm)
}

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 checkValidity(vm varMap) error {
	for k, v := range vm {
		typ := v.Type

		if len(typ) == 0 {
			return x.Errorf("Type of variable %v not specified", k)
		}

		// Ensure value is not nil if the variable is required.
		if typ[len(typ)-1] == '!' {
			if v.Value == "" {
				return x.Errorf("Variable %v should be initialised", k)
			}
			typ = typ[:len(typ)-1]
		}

		// Type check the values.
		if v.Value != "" {
			switch typ {
			case "int":
				{
					if _, err := strconv.ParseInt(v.Value, 0, 64); err != nil {
						return x.Wrapf(err, "Expected an int but got %v", v.Value)
					}
				}
			case "float":
				{
					if _, err := strconv.ParseFloat(v.Value, 64); err != nil {
						return x.Wrapf(err, "Expected a float but got %v", v.Value)
					}
				}
			case "bool":
				{
					if _, err := strconv.ParseBool(v.Value); err != nil {
						return x.Wrapf(err, "Expected a bool but got %v", v.Value)
					}
				}
			case "string": // Value is a valid string. No checks required.
			default:
				return x.Errorf("Type %v not supported", typ)
			}
		}
	}

	return nil
}