// WaitUntilIndexState comes to desired `state`,
// retry for every `period` mS until `timeout` mS.
func WaitUntilIndexState(
	client *qclient.GsiClient, defnIDs []uint64,
	state c.IndexState, period, timeout time.Duration) ([]c.IndexState, error) {

	expired := time.After(timeout * time.Millisecond)
	states := make([]c.IndexState, len(defnIDs))
	pending := len(defnIDs)
	for {
		select {
		case <-expired:
			return nil, errors.New("timeout")

		default:
		}
		for i, defnID := range defnIDs {
			if states[i] != state {
				st, err := client.IndexState(defnID)
				if err != nil {
					return nil, err
				} else if st == state {
					states[i] = state
					pending--
					continue
				}
			}
		}
		if pending == 0 {
			return states, nil
		}
		time.Sleep(period * time.Millisecond)
	}
}

func sessionConsistency(
	client *qclient.GsiClient,
	defnID uint64,
	equals []common.SecondaryKey, synch chan bool) {

	go func() {
		fmt.Println("Scan: SessionConsistency ...")
		client.Lookup(
			uint64(defnID), equals, false, 10, common.SessionConsistency, nil,
			func(res qclient.ResponseReader) bool {
				if res.Error() != nil {
					log.Fatalf("Error: %v", res)
				} else if skeys, pkeys, err := res.GetEntries(); err != nil {
					log.Fatalf("Error: %v", err)
				} else {
					for i, pkey := range pkeys {
						fmt.Printf("    %v ... %v\n", skeys[i], string(pkey))
					}
				}
				return true
			})
		fmt.Println("Scan: SessionConsistency ... ok\n")
		synch <- true
	}()
}

func GetDefnID(client *qc.GsiClient, bucket, indexName string) (defnID uint64, ok bool) {
	indexes, err := client.Refresh()
	tc.HandleError(err, "Error while listing the indexes")
	for _, index := range indexes {
		defn := index.Definition
		if defn.Bucket == bucket && defn.Name == indexName {
			return uint64(index.Definition.DefnId), true
		}
	}
	return uint64(c.IndexDefnId(0)), false
}

func IndexExistsWithClient(indexName, bucketName, server string, client *qc.GsiClient) bool {
	indexes, err := client.Refresh()
	tc.HandleError(err, "Error while listing the secondary indexes")
	for _, index := range indexes {
		defn := index.Definition
		if defn.Name == indexName && defn.Bucket == bucketName {
			log.Printf("Index found:  %v", indexName)
			return true
		}
	}
	return false
}

// Creates an index and waits for it to become active
func CreateSecondaryIndex(
	indexName, bucketName, server, whereExpr string, indexFields []string, isPrimary bool, with []byte,
	skipIfExists bool, indexActiveTimeoutSeconds int64, client *qc.GsiClient) error {

	if client == nil {
		c, e := CreateClient(server, "2itest")
		if e != nil {
			return e
		}
		client = c
		defer client.Close()
	}

	indexExists := IndexExistsWithClient(indexName, bucketName, server, client)
	if skipIfExists == true && indexExists == true {
		return nil
	}
	var secExprs []string
	if isPrimary == false {
		for _, indexField := range indexFields {
			expr, err := n1ql.ParseExpression(indexField)
			if err != nil {
				log.Printf("Creating index %v. Error while parsing the expression (%v) : %v", indexName, indexField, err)
			}

			secExprs = append(secExprs, expression.NewStringer().Visit(expr))
		}
	}
	using := "gsi"
	exprType := "N1QL"
	partnExp := ""

	start := time.Now()
	defnID, err := client.CreateIndex(indexName, bucketName, using, exprType, partnExp, whereExpr, secExprs, isPrimary, with)
	if err == nil {
		log.Printf("Created the secondary index %v. Waiting for it become active", indexName)
		e := WaitTillIndexActive(defnID, client, indexActiveTimeoutSeconds)
		if e != nil {
			return e
		} else {
			elapsed := time.Since(start)
			tc.LogPerfStat("CreateAndBuildIndex", elapsed)
			return nil
		}
	}

	return err
}

// GetIndex for bucket/indexName.
func GetIndex(
	client *qclient.GsiClient,
	bucket, indexName string) (*mclient.IndexMetadata, bool) {

	indexes, err := client.Refresh()
	if err != nil {
		logging.Fatalf("%v\n", err)
		os.Exit(1)
	}
	for _, index := range indexes {
		defn := index.Definition
		if defn.Bucket == bucket && defn.Name == indexName {
			return index, true
			//return uint64(index.Definition.DefnId), true
		}
	}
	return nil, false
}

func RangeWithClient(indexName, bucketName, server string, low, high []interface{}, inclusion uint32,
	distinct bool, limit int64, consistency c.Consistency, vector *qc.TsConsistency, client *qc.GsiClient) (tc.ScanResponse, error) {
	var scanErr error
	scanErr = nil
	defnID, _ := GetDefnID(client, bucketName, indexName)
	scanResults := make(tc.ScanResponse)

	start := time.Now()
	connErr := client.Range(
		defnID, c.SecondaryKey(low), c.SecondaryKey(high), qc.Inclusion(inclusion), distinct, limit,
		consistency, vector,
		func(response qc.ResponseReader) bool {
			if err := response.Error(); err != nil {
				scanErr = err
				return false
			} else if skeys, pkeys, err := response.GetEntries(); err != nil {
				scanErr = err
				return false
			} else {
				for i, skey := range skeys {
					primaryKey := string(pkeys[i])
					if _, keyPresent := scanResults[primaryKey]; keyPresent {
						// Duplicate primary key found
						tc.HandleError(err, "Duplicate primary key found in the scan results: "+primaryKey)
					} else {
						scanResults[primaryKey] = skey
					}
				}
				return true
			}
			return false
		})
	elapsed := time.Since(start)

	if connErr != nil {
		log.Printf("Connection error in Scan occured: %v", connErr)
		return scanResults, connErr
	} else if scanErr != nil {
		return scanResults, scanErr
	}

	tc.LogPerfStat("Range", elapsed)
	return scanResults, nil
}

func DropSecondaryIndexWithClient(indexName, bucketName, server string, client *qc.GsiClient) error {
	log.Printf("Dropping the secondary index %v", indexName)
	indexes, err := client.Refresh()
	tc.HandleError(err, "Error while listing the secondary indexes")
	for _, index := range indexes {
		defn := index.Definition
		if (defn.Name == indexName) && (defn.Bucket == bucketName) {
			start := time.Now()
			e := client.DropIndex(uint64(defn.DefnId))
			elapsed := time.Since(start)
			if e == nil {
				log.Printf("Index dropped")
				tc.LogPerfStat("DropIndex", elapsed)
			} else {
				return e
			}
		}
	}
	return nil
}

// Creates an index and DOES NOT wait for it to become active
func CreateSecondaryIndexAsync(
	indexName, bucketName, server, whereExpr string, indexFields []string, isPrimary bool, with []byte,
	skipIfExists bool, client *qc.GsiClient) error {

	if client == nil {
		c, e := CreateClient(server, "2itest")
		if e != nil {
			return e
		}
		client = c
		defer client.Close()
	}

	indexExists := IndexExistsWithClient(indexName, bucketName, server, client)
	if skipIfExists == true && indexExists == true {
		return nil
	}
	var secExprs []string
	if isPrimary == false {
		for _, indexField := range indexFields {
			expr, err := n1ql.ParseExpression(indexField)
			if err != nil {
				log.Printf("Creating index %v. Error while parsing the expression (%v) : %v", indexName, indexField, err)
			}

			secExprs = append(secExprs, expression.NewStringer().Visit(expr))
		}
	}
	using := "gsi"
	exprType := "N1QL"
	partnExp := ""

	_, err := client.CreateIndex(indexName, bucketName, using, exprType, partnExp, whereExpr, secExprs, isPrimary, with)
	if err == nil {
		log.Printf("Created the secondary index %v", indexName)
		return nil
	}
	return err
}

func WaitTillIndexActive(defnID uint64, client *qc.GsiClient, indexActiveTimeoutSeconds int64) error {
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= float64(indexActiveTimeoutSeconds) {
			err := errors.New(fmt.Sprintf("Index did not become active after %d seconds", indexActiveTimeoutSeconds))
			return err
		}
		state, e := client.IndexState(defnID)
		log.Printf("Index state is %v", state)
		if e != nil {
			log.Printf("Error while fetching index state for defnID %v", defnID)
			return e
		}

		if state == c.INDEX_STATE_ACTIVE {
			log.Printf("Index is now active")
			return nil
		} else {
			time.Sleep(1 * time.Second)
		}
	}
	return nil
}

func n1qlError(client *qclient.GsiClient, err error) errors.Error {
	return errors.NewError(err, client.DescribeError(err))
}

func RunJob(client *qclient.GsiClient, job *Job, aggrQ chan *JobResult) {
	var err error
	var rows int64

	spec := job.spec
	result := job.result
	if result != nil {
		result.Id = spec.Id
	}

	errFn := func(e string) {
		fmt.Printf("REQ:%d scan error occured: %s\n", spec.Id, e)
		if result != nil {
			platform.AddUint64(&result.ErrorCount, 1)
		}
	}

	callb := func(res qclient.ResponseReader) bool {
		if res.Error() != nil {
			errFn(res.Error().Error())
			return false
		} else {
			_, pkeys, err := res.GetEntries()
			if err != nil {
				errFn(err.Error())
				return false
			}

			rows += int64(len(pkeys))
		}

		return true
	}

	var cons c.Consistency
	if spec.Consistency {
		cons = c.SessionConsistency
	} else {
		cons = c.AnyConsistency
	}

	startTime := time.Now()
	switch spec.Type {
	case "All":
		err = client.ScanAll(spec.DefnId, spec.Limit, cons, nil, callb)
	case "Range":
		err = client.Range(spec.DefnId, spec.Low, spec.High,
			qclient.Inclusion(spec.Inclusion), false, spec.Limit, cons, nil, callb)
	case "Lookup":
		err = client.Lookup(spec.DefnId, spec.Lookups, false,
			spec.Limit, cons, nil, callb)
	}

	if err != nil {
		errFn(err.Error())
	}

	dur := time.Now().Sub(startTime)

	if result != nil {
		aggrQ <- &JobResult{
			job:  job,
			dur:  dur.Nanoseconds(),
			rows: rows,
		}
	}
}

// HandleCommand after parsing it with ParseArgs().
func HandleCommand(
	client *qclient.GsiClient,
	cmd *Command,
	verbose bool,
	w io.Writer) (err error) {

	iname, bucket, limit := cmd.IndexName, cmd.Bucket, cmd.Limit
	low, high, equal, incl := cmd.Low, cmd.High, cmd.Equal, cmd.Inclusion
	cons := cmd.Consistency

	indexes, err := client.Refresh()

	entries := 0
	callb := func(res qclient.ResponseReader) bool {
		if res.Error() != nil {
			fmt.Fprintln(w, "Error: ", res)
		} else if skeys, pkeys, err := res.GetEntries(); err != nil {
			fmt.Fprintln(w, "Error: ", err)
		} else {
			if verbose == false {
				for i, pkey := range pkeys {
					fmt.Fprintf(w, "%v ... %v\n", skeys[i], string(pkey))
				}
			}
			entries += len(pkeys)
		}
		return true
	}

	switch cmd.OpType {
	case "nodes":
		fmt.Fprintln(w, "List of nodes:")
		nodes, err := client.Nodes()
		if err != nil {
			return err
		}
		for _, n := range nodes {
			fmsg := "    {%v, %v, %q}\n"
			fmt.Fprintf(w, fmsg, n.Adminport, n.Queryport, n.Status)
		}

	case "list":
		time.Sleep(2 * time.Second)
		indexes, err = client.Refresh()
		if err != nil {
			return err
		}
		fmt.Fprintln(w, "List of indexes:")
		for _, index := range indexes {
			printIndexInfo(w, index)
		}

	case "create":
		var defnID uint64
		if len(cmd.SecStrs) == 0 && !cmd.IsPrimary || cmd.IndexName == "" {
			return fmt.Errorf("createIndex(): required fields missing")
		}
		defnID, err = client.CreateIndex(
			iname, bucket, cmd.Using, cmd.ExprType,
			cmd.PartnStr, cmd.WhereStr, cmd.SecStrs, cmd.IsPrimary,
			[]byte(cmd.With))
		if err == nil {
			fmt.Fprintf(w, "Index created: %v with %q\n", defnID, cmd.With)
		}

	case "build":
		defnIDs := make([]uint64, 0, len(cmd.Bindexes))
		for _, bindex := range cmd.Bindexes {
			v := strings.Split(bindex, ":")
			if len(v) < 0 {
				return fmt.Errorf("invalid index specified : %v", bindex)
			}
			bucket, iname = v[0], v[1]
			index, ok := GetIndex(client, bucket, iname)
			if ok {
				defnIDs = append(defnIDs, uint64(index.Definition.DefnId))
			} else {
				err = fmt.Errorf("index %v/%v unknown", bucket, iname)
				break
			}
		}
		if err == nil {
			err = client.BuildIndexes(defnIDs)
			fmt.Fprintf(w, "Index building for: %v\n", defnIDs)
		}

	case "drop":
		index, ok := GetIndex(client, cmd.Bucket, cmd.IndexName)
		if !ok {
			return fmt.Errorf("invalid index specified : %v", cmd.IndexName)
		}
		err = client.DropIndex(uint64(index.Definition.DefnId))
		if err == nil {
			fmt.Fprintf(w, "Index dropped %v/%v\n", bucket, iname)
		} else {
			err = fmt.Errorf("index %v/%v drop failed", bucket, iname)
			break
		}

//.........这里部分代码省略.........

func n1qlError(client *qclient.GsiClient, err error) errors.Error {
	if err.Error() == c.ErrScanTimedOut.Error() {
		return errors.NewCbIndexScanTimeoutError(err)
	}
	return errors.NewError(err, client.DescribeError(err))
}