// CreatePrimaryIndex implements datastore.Indexer{} interface. Create or
// return a primary index on this keyspace
func (gsi *gsiKeyspace) CreatePrimaryIndex(
	requestId, name string, with value.Value) (datastore.PrimaryIndex, errors.Error) {

	var withJSON []byte
	var err error
	if with != nil {
		if withJSON, err = with.MarshalJSON(); err != nil {
			return nil, errors.NewError(err, "GSI error marshalling WITH clause")
		}
	}
	defnID, err := gsi.gsiClient.CreateIndex(
		name,
		gsi.keyspace,       /*bucket-name*/
		string(c.ForestDB), /*using, by default always forestdb*/
		"N1QL",             /*exprType*/
		"",                 /*partnStr*/
		"",                 /*whereStr*/
		nil,                /*secStrs*/
		true,               /*isPrimary*/
		withJSON)
	if err != nil {
		return nil, errors.NewError(err, "GSI CreatePrimaryIndex()")
	}
	// refresh to get back the newly created index.
	if err := gsi.Refresh(); err != nil {
		return nil, err
	}
	index, errr := gsi.IndexById(defnID2String(defnID))
	if errr != nil {
		return nil, errr
	}
	return index.(datastore.PrimaryIndex), nil
}

// BuildIndexes implements datastore.Indexer{} interface.
func (gsi *gsiKeyspace) BuildIndexes(requestId string, names ...string) errors.Error {
	defnIDs := make([]uint64, len(names))
	for i, name := range names {
		index, err := gsi.IndexByName(name)
		if err != nil {
			return errors.NewError(err, "BuildIndexes")
		}
		defnIDs[i] = string2defnID(index.Id())
	}
	err := gsi.gsiClient.BuildIndexes(defnIDs)
	if err != nil {
		return errors.NewError(err, "BuildIndexes")
	}
	return nil
}

// for metadata-provider.
func newSecondaryIndexFromMetaData(
	gsi *gsiKeyspace,
	imd *mclient.IndexMetadata) (si *secondaryIndex, err errors.Error) {

	if len(imd.Instances) < 1 {
		return nil, errors.NewError(nil, "no instance are created by GSI")
	}
	instn, indexDefn := imd.Instances[0], imd.Definition
	defnID := uint64(indexDefn.DefnId)
	si = &secondaryIndex{
		gsi:       gsi,
		bucketn:   indexDefn.Bucket,
		name:      indexDefn.Name,
		defnID:    defnID,
		isPrimary: indexDefn.IsPrimary,
		using:     indexDefn.Using,
		partnExpr: indexDefn.PartitionKey,
		secExprs:  indexDefn.SecExprs,
		whereExpr: indexDefn.WhereExpr,
		state:     gsi2N1QLState[instn.State],
		err:       instn.Error,
		deferred:  indexDefn.Deferred,
	}
	return si, nil
}

func makeResponsehandler(
	client *qclient.GsiClient, conn *datastore.IndexConnection) qclient.ResponseHandler {

	entryChannel := conn.EntryChannel()
	stopChannel := conn.StopChannel()

	return func(data qclient.ResponseReader) bool {

		if err := data.Error(); err != nil {
			conn.Error(n1qlError(client, err))
			return false
		}
		skeys, pkeys, err := data.GetEntries()
		if err == nil {
			for i, skey := range skeys {
				// Primary-key is mandatory.
				e := &datastore.IndexEntry{
					PrimaryKey: string(pkeys[i]),
				}
				e.EntryKey = skey2Values(skey)

				fmsg := "current enqueued length: %d (max %d)\n"
				l.Tracef(fmsg, len(entryChannel), cap(entryChannel))
				select {
				case entryChannel <- e:
				case <-stopChannel:
					return false
				}
			}
			return true
		}
		conn.Error(errors.NewError(nil, err.Error()))
		return false
	}
}

// NewGSIIndexer manage new set of indexes under namespace->keyspace,
// also called as, pool->bucket.
// will return an error when,
// - GSI cluster is not available.
// - network partitions / errors.
func NewGSIIndexer(
	clusterURL, namespace, keyspace string) (datastore.Indexer, errors.Error) {

	l.SetLogLevel(l.Info)

	gsi := &gsiKeyspace{
		clusterURL:     clusterURL,
		namespace:      namespace,
		keyspace:       keyspace,
		indexes:        make(map[uint64]*secondaryIndex), // defnID -> index
		primaryIndexes: make(map[uint64]*secondaryIndex),
	}
	gsi.logPrefix = fmt.Sprintf("GSIC[%s; %s]", namespace, keyspace)

	// get the singleton-client
	client, err := getSingletonClient(clusterURL)
	if err != nil {
		l.Errorf("%v GSI instantiation failed: %v", gsi.logPrefix, err)
		return nil, errors.NewError(err, "GSI client instantiation failed")
	}
	gsi.gsiClient = client
	// refresh indexes for this service->namespace->keyspace
	if err := gsi.Refresh(); err != nil {
		l.Errorf("%v Refresh() failed: %v", gsi.logPrefix, err)
		return nil, err
	}
	l.Debugf("%v instantiated ...", gsi.logPrefix)
	return gsi, nil
}

func Run(mockServer *server.Server, q string) ([]interface{}, []errors.Error, errors.Error) {
	var metrics value.Tristate
	scanConfiguration := &scanConfigImpl{}

	base := server.NewBaseRequest(q, nil, nil, nil, "json", 0, value.FALSE, metrics, value.TRUE, scanConfiguration, "", nil)

	mr := &MockResponse{
		results: []interface{}{}, warnings: []errors.Error{}, done: make(chan bool),
	}

	query := &MockQuery{
		BaseRequest: *base,
		response:    mr,
	}

	select {
	case mockServer.Channel() <- query:
		// Wait until the request exits.
		<-query.CloseNotify()
	default:
		// Timeout.
		return nil, nil, errors.NewError(nil, "Query timed out")
	}

	// wait till all the results are ready
	<-mr.done
	return mr.results, mr.warnings, mr.err
}

// Synchronise gsi client with the servers and refresh the indexes list.
func (gsi *gsiKeyspace) SyncRefresh() errors.Error {
	err := gsi.gsiClient.Sync()
	if err != nil {
		return errors.NewError(err, "GSI SyncRefresh()")
	}
	return gsi.Refresh()
}

// get cluster info and refresh ns-server data.
func getClusterInfo(
	cluster string, pooln string) (*c.ClusterInfoCache, errors.Error) {

	clusterURL, err := c.ClusterAuthUrl(cluster)
	if err != nil {
		return nil, errors.NewError(err, fmt.Sprintf("ClusterAuthUrl() failed"))
	}
	cinfo, err := c.NewClusterInfoCache(clusterURL, pooln)
	if err != nil {
		return nil, errors.NewError(err, fmt.Sprintf("ClusterInfo() failed"))
	}
	if err := cinfo.Fetch(); err != nil {
		msg := fmt.Sprintf("Fetch ClusterInfo() failed")
		return nil, errors.NewError(err, msg)
	}
	return cinfo, nil
}

// CreateIndex implements datastore.Indexer{} interface. Create a secondary
// index on this keyspace
func (gsi *gsiKeyspace) CreateIndex(
	requestId, name string, seekKey, rangeKey expression.Expressions,
	where expression.Expression, with value.Value) (
	datastore.Index, errors.Error) {

	var partnStr string
	if seekKey != nil && len(seekKey) > 0 {
		partnStr = expression.NewStringer().Visit(seekKey[0])
	}

	var whereStr string
	if where != nil {
		whereStr = expression.NewStringer().Visit(where)
	}

	secStrs := make([]string, len(rangeKey))
	for i, key := range rangeKey {
		s := expression.NewStringer().Visit(key)
		secStrs[i] = s
	}

	var withJSON []byte
	var err error
	if with != nil {
		if withJSON, err = with.MarshalJSON(); err != nil {
			return nil, errors.NewError(err, "GSI error marshalling WITH clause")
		}
	}
	defnID, err := gsi.gsiClient.CreateIndex(
		name,
		gsi.keyspace,       /*bucket-name*/
		string(c.ForestDB), /*using, by default always forestdb*/
		"N1QL",             /*exprType*/
		partnStr, whereStr, secStrs,
		false, /*isPrimary*/
		withJSON)
	if err != nil {
		return nil, errors.NewError(err, "GSI CreateIndex()")
	}
	// refresh to get back the newly created index.
	if err := gsi.Refresh(); err != nil {
		return nil, err
	}
	return gsi.IndexById(defnID2String(defnID))
}

// Drop implement Index{} interface.
func (si *secondaryIndex) Drop(requestId string) errors.Error {
	if si == nil {
		return ErrorIndexEmpty
	}
	if err := si.gsi.gsiClient.DropIndex(si.defnID); err != nil {
		return errors.NewError(err, "GSI Drop()")
	}
	si.gsi.delIndex(si.Id())
	return nil
}

func (this *Context) Recover() {
	err := recover()
	if err != nil {
		buf := make([]byte, 1<<16)
		n := runtime.Stack(buf, false)
		s := string(buf[0:n])
		logging.Severep("", logging.Pair{"panic", err},
			logging.Pair{"stack", s})
		os.Stderr.WriteString(s)
		os.Stderr.Sync()

		switch err := err.(type) {
		case error:
			this.Fatal(errors.NewError(err, fmt.Sprintf("Panic: %v", err)))
		default:
			this.Fatal(errors.NewError(nil, fmt.Sprintf("Panic: %v", err)))
		}
	}
}

// helper function to determine the external IP address of a query node -
// used to create a name for the query node in NewQueryNode function.
func ExternalIP() (string, errors.Error) {
	ifaces, err := net.Interfaces()
	if err != nil {
		return "", errors.NewError(err, "")
	}
	result := ""

	for _, iface := range ifaces {
		if iface.Flags&net.FlagUp == 0 {
			continue // interface down
		}
		if iface.Flags&net.FlagLoopback != 0 {
			continue // loopback interface
		}
		addrs, err := iface.Addrs()
		if err != nil {
			return "", errors.NewError(err, "")
		}
		for _, addr := range addrs {
			var ip net.IP
			switch v := addr.(type) {
			case *net.IPNet:
				ip = v.IP
			case *net.IPAddr:
				ip = v.IP
			}
			if ip == nil || ip.IsLoopback() {
				continue
			}
			ip = ip.To4()
			if ip == nil {
				continue // not an ipv4 address
			}
			result = ip.String()
			if strings.HasPrefix(result, "192") {
				return result, nil
			}
		}
	}
	return result, errors.NewError(nil, "Not connected to the network")
}

func GetKeyspace(namespace, keyspace string) (Keyspace, errors.Error) {
	datastore := GetDatastore()
	if datastore == nil {
		return nil, errors.NewError(nil, "Datastore not set.")
	}

	ns, err := datastore.NamespaceByName(namespace)
	if err != nil {
		return nil, err
	}

	return ns.KeyspaceByName(keyspace)
}

func (this *base) requireKey(item value.AnnotatedValue, context *Context) (string, bool) {
	mv := item.GetAttachment("meta")
	if mv == nil {
		context.Error(errors.NewError(nil, "Unable to find meta."))
		return "", false
	}

	meta := mv.(map[string]interface{})
	key, ok := meta["id"]
	if !ok {
		context.Error(errors.NewError(nil, "Unable to find key."))
		return "", false
	}

	act := value.NewValue(key).Actual()
	switch act := act.(type) {
	case string:
		return act, true
	default:
		e := errors.NewError(nil, fmt.Sprintf("Unable to process non-string key %v of type %T.", act, act))
		context.Error(e)
		return "", false
	}
}

// IndexById implements datastore.Indexer{} interface. Find an index on this
// keyspace using the index's id.
func (gsi *gsiKeyspace) IndexById(id string) (datastore.Index, errors.Error) {
	gsi.rw.RLock()
	defer gsi.rw.RUnlock()
	defnID := string2defnID(id)
	index, ok := gsi.indexes[defnID]
	if !ok {
		index, ok = gsi.primaryIndexes[defnID]
		if !ok {
			errmsg := fmt.Sprintf("GSI index id %v not found.", id)
			err := errors.NewError(nil, errmsg)
			return nil, err
		}
	}
	l.Debugf("%v IndexById %v = %v", gsi.logPrefix, id, index)
	return index, nil
}

func (this *Explain) RunOnce(context *Context, parent value.Value) {
	this.once.Do(func() {
		defer context.Recover()       // Recover from any panic
		defer close(this.itemChannel) // Broadcast that I have stopped
		defer this.notify()           // Notify that I have stopped

		bytes, err := json.Marshal(this.plan)
		if err != nil {
			context.Fatal(errors.NewError(err, "Failed to marshal JSON."))
			return
		}

		value := value.NewAnnotatedValue(bytes)
		this.sendItem(value)

	})
}

// IndexByName implements datastore.Indexer{} interface. Find an index on
// this keyspace using the index's name.
func (gsi *gsiKeyspace) IndexByName(name string) (datastore.Index, errors.Error) {
	gsi.rw.RLock()
	defer gsi.rw.RUnlock()

	for _, index := range gsi.indexes {
		if index.Name() == name {
			return index, nil
		}
	}
	for _, index := range gsi.primaryIndexes {
		if index.Name() == name {
			return index, nil
		}
	}
	err := errors.NewError(nil, fmt.Sprintf("GSI index %v not found.", name))
	return nil, err
}

func (this *InferKeyspace) RunOnce(context *Context, parent value.Value) {
	this.once.Do(func() {
		defer context.Recover()       // Recover from any panic
		defer close(this.itemChannel) // Broadcast that I have stopped
		defer this.notify()           // Notify that I have stopped

		conn := datastore.NewValueConnection(context)
		defer notifyConn(conn.StopChannel())

		var duration time.Duration
		timer := time.Now()
		defer context.AddPhaseTime("InferKeySpace", time.Since(timer)-duration)

		infer, err := context.Datastore().Inferencer(this.plan.Node().Using())
		if err != nil {
			context.Error(errors.NewError(err, "Failed to get Inferencer"))
			return
		}
		go infer.InferKeyspace(this.plan.Keyspace(), this.plan.Node().With(), conn)

		var val value.Value

		ok := true
		for ok {
			select {
			case <-this.stopChannel:
				return
			default:
			}

			select {
			case val, ok = <-conn.ValueChannel():
				if ok {
					t := time.Now()
					ok = this.sendItem(value.NewAnnotatedValue(val))
					duration += time.Since(t)
				}
			case <-this.stopChannel:
				return
			}
		}
	})
}

/*
Returns all required privileges.
*/
func subqueryPrivileges(exprs expression.Expressions) (datastore.Privileges, errors.Error) {
	subqueries, err := expression.ListSubqueries(exprs, false)
	if err != nil {
		return nil, errors.NewError(err, "")
	}

	privileges := datastore.NewPrivileges()
	for _, s := range subqueries {
		sub := s.(*Subquery)
		sp, e := sub.Select().Privileges()
		if e != nil {
			return nil, e
		}

		privileges.Add(sp)
	}

	return privileges, nil
}

func GetKeyspace(namespace, keyspace string) (Keyspace, errors.Error) {
	var datastore Datastore

	if namespace == "#system" {
		datastore = GetSystemstore()
	} else {
		datastore = GetDatastore()
	}

	if datastore == nil {
		return nil, errors.NewError(nil, "Datastore not set.")
	}

	ns, err := datastore.NamespaceByName(namespace)
	if err != nil {
		return nil, err
	}

	return ns.KeyspaceByName(keyspace)
}