/*
Qualify all identifiers for the parent expression. Checks for
duplicate aliases.
*/
func (this *SubqueryTerm) Formalize(parent *expression.Formalizer) (f *expression.Formalizer, err error) {
	err = this.subquery.Formalize()
	if err != nil {
		return
	}

	alias := this.Alias()
	if alias == "" {
		err = errors.NewError(nil, "FROM term must have a name or alias.")
		return
	}

	_, ok := parent.Allowed.Field(alias)
	if ok {
		err = errors.NewError(nil, fmt.Sprintf("Duplicate subquery alias %s.", alias))
		return nil, err
	}

	allowed := value.NewScopeValue(make(map[string]interface{}), parent.Allowed)
	allowed.SetField(alias, alias)

	f = expression.NewFormalizer()
	f.Keyspace = alias
	f.Allowed = allowed
	return
}

func (this *IntersectScan) processKey(item value.AnnotatedValue, context *Context) bool {
	m := item.GetAttachment("meta")
	meta, ok := m.(map[string]interface{})
	if !ok {
		context.Error(errors.NewError(nil,
			fmt.Sprintf("Missing or invalid meta %v of type %T.", m, m)))
		return false
	}

	k := meta["id"]
	key, ok := k.(string)
	if !ok {
		context.Error(errors.NewError(nil,
			fmt.Sprintf("Missing or invalid primary key %v of type %T.", k, k)))
		return false
	}

	count := this.counts[key]
	this.counts[key] = count + 1

	if count+1 == len(this.scans) {
		delete(this.values, key)
		return this.sendItem(item)
	}

	if count == 0 {
		this.values[key] = item
	}

	return true
}

func (this *ValueScan) 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

		pairs := this.plan.Values()
		for _, pair := range pairs {
			key, err := pair.Key.Evaluate(parent, context)
			if err != nil {
				context.Error(errors.NewError(err, "Error evaluating VALUES."))
				return
			}

			val, err := pair.Value.Evaluate(parent, context)
			if err != nil {
				context.Error(errors.NewError(err, "Error evaluating VALUES."))
				return
			}

			av := value.NewAnnotatedValue(nil)
			av.SetAttachment("key", key)
			av.SetAttachment("value", val)

			if !this.sendItem(av) {
				return
			}
		}
	})
}

func (this *InitialProject) processTerms(item value.AnnotatedValue, context *Context) bool {
	n := len(this.plan.Terms())

	var f map[string]interface{}
	if item.Type() == value.OBJECT {
		f = item.Copy().Fields()
	}

	if f == nil {
		f = make(map[string]interface{}, n)
	}

	pv := value.NewAnnotatedValue(f)
	pv.SetAttachments(item.Attachments())

	p := value.NewValue(make(map[string]interface{}, n+32))
	pv.SetAttachment("projection", p)

	for _, term := range this.plan.Terms() {
		if term.Result().Alias() != "" {
			v, err := term.Result().Expression().Evaluate(item, context)
			if err != nil {
				context.Error(errors.NewError(err, "Error evaluating projection."))
				return false
			}

			p.SetField(term.Result().Alias(), v)

			// Explicit aliases override data
			if term.Result().As() != "" {
				pv.SetField(term.Result().As(), v)
			}
		} else {
			// Star
			starval := item.GetValue()
			if term.Result().Expression() != nil {
				var err error
				starval, err = term.Result().Expression().Evaluate(item, context)
				if err != nil {
					context.Error(errors.NewError(err, "Error evaluating projection."))
					return false
				}
			}

			// Latest star overwrites previous star
			switch sa := starval.Actual().(type) {
			case map[string]interface{}:
				for k, v := range sa {
					p.SetField(k, v)
				}
			}
		}
	}

	return this.sendItem(pv)
}

func (this *Order) Less(i, j int) bool {
	v1 := this.values[i]
	v2 := this.values[j]

	var ev1, ev2 value.Value
	var c int
	var e error

	for i, term := range this.plan.Terms() {
		s := strconv.Itoa(i)

		sv1 := v1.GetAttachment(s)
		switch sv1 := sv1.(type) {
		case value.Value:
			ev1 = sv1
		default:
			ev1, e = term.Expression().Evaluate(v1, this.context)
			if e != nil {
				this.context.Error(errors.NewError(e, "Error evaluating ORDER BY."))
				return false
			}

			v1.SetAttachment(s, ev1)
		}

		sv2 := v2.GetAttachment(s)
		switch sv2 := sv2.(type) {
		case value.Value:
			ev2 = sv2
		default:
			ev2, e = term.Expression().Evaluate(v2, this.context)
			if e != nil {
				this.context.Error(errors.NewError(e, "Error evaluating ORDER BY."))
				return false
			}

			v2.SetAttachment(s, ev2)
		}

		c = ev1.Collate(ev2)

		if c == 0 {
			continue
		} else if term.Descending() {
			return c > 0
		} else {
			return c < 0
		}
	}

	return false
}

func (this *Merge) processMatch(item value.AnnotatedValue,
	context *Context, update, delete, insert Operator) bool {
	kv, e := this.plan.Key().Evaluate(item, context)
	if e != nil {
		context.Error(errors.NewError(e, "Error evaluatating MERGE key."))
		return false
	}

	ka := kv.Actual()
	k, ok := ka.(string)
	if !ok {
		context.Error(errors.NewError(nil,
			fmt.Sprintf("Invalid MERGE key %v of type %T.", ka, ka)))
		return false
	}

	bvs, err := this.plan.Keyspace().Fetch([]string{k})
	if err != nil {
		context.Error(err)
		return false
	}

	if len(bvs) > 0 {
		bv := bvs[0].Value

		// Matched; join source and target
		if update != nil {
			item.SetAttachment("target", bv)
		}

		abv := value.NewAnnotatedValue(bv)
		item.SetField(this.plan.KeyspaceRef().Alias(), abv)

		// Perform UPDATE and/or DELETE
		if update != nil {
			update.Input().ItemChannel() <- item
		}

		if delete != nil {
			delete.Input().ItemChannel() <- item
		}
	} else {
		// Not matched; INSERT
		if insert != nil {
			insert.Input().ItemChannel() <- item
		}
	}

	return true
}

func (this *builder) VisitKeyspaceTerm(node *algebra.KeyspaceTerm) (interface{}, error) {
	node.SetDefaultNamespace(this.namespace)
	keyspace, err := this.getTermKeyspace(node)
	if err != nil {
		return nil, err
	}

	if node.Keys() != nil {
		scan := NewKeyScan(node.Keys())
		this.children = append(this.children, scan)
	} else {
		if this.subquery {
			return nil, errors.NewError(nil, fmt.Sprintf(
				"FROM in subquery must use KEYS clause: FROM %s.",
				node.Keyspace()))
		}

		scan, err := this.selectScan(keyspace, node)
		if err != nil {
			return nil, err
		}

		this.children = append(this.children, scan)
	}

	fetch := NewFetch(keyspace, node)
	this.subChildren = append(this.subChildren, fetch)
	return nil, nil
}

func (this *KeyScan) 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

		keys, e := this.plan.Keys().Evaluate(parent, context)
		if e != nil {
			context.Error(errors.NewError(e, "Error evaluating KEYS."))
			return
		}

		actuals := keys.Actual()
		switch actuals.(type) {
		case []interface{}:
		case nil:
			actuals = []interface{}(nil)
		default:
			actuals = []interface{}{actuals}
		}

		acts := actuals.([]interface{})

		for _, key := range acts {
			cv := value.NewScopeValue(make(map[string]interface{}), parent)
			av := value.NewAnnotatedValue(cv)
			av.SetAttachment("meta", map[string]interface{}{"id": key})
			if !this.sendItem(av) {
				return
			}
		}
	})
}

func Run(mockServer *server.Server, q string) ([]interface{}, []errors.Error, errors.Error) {

	var metrics value.Tristate
	base := server.NewBaseRequest(q, nil, nil, nil, "json", value.FALSE, metrics, value.TRUE, nil, "", 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
}

func (this *FinalGroup) processItem(item value.AnnotatedValue, context *Context) bool {
	// Generate the group key
	var gk string
	if len(this.plan.Keys()) > 0 {
		var e error
		gk, e = groupKey(item, this.plan.Keys(), context)
		if e != nil {
			context.Error(errors.NewError(e, "Error evaluating GROUP key."))
			return false
		}
	}

	// Get or seed the group value
	gv := this.groups[gk]
	if gv != nil {
		context.Error(errors.NewError(nil, "Duplicate final GROUP."))
		return false
	}

	gv = item
	this.groups[gk] = gv

	// Compute final aggregates
	aggregates := gv.GetAttachment("aggregates")
	switch aggregates := aggregates.(type) {
	case map[string]value.Value:
		for _, agg := range this.plan.Aggregates() {
			v, e := agg.ComputeFinal(aggregates[agg.String()], context)
			if e != nil {
				context.Error(errors.NewError(
					e, "Error updating GROUP value."))
				return false
			}

			aggregates[agg.String()] = v
		}

		return true
	default:
		context.Error(errors.NewError(nil, fmt.Sprintf(
			"Invalid or missing aggregates of type %T.", aggregates)))
		return false
	}
}

func (this *Limit) beforeItems(context *Context, parent value.Value) bool {
	val, e := this.plan.Expression().Evaluate(parent, context)
	if e != nil {
		context.Error(errors.NewError(e, "Error evaluating LIMIT."))
		return false
	}

	actual := val.Actual()
	switch actual := actual.(type) {
	case float64:
		if math.Trunc(actual) == actual {
			this.limit = int64(actual)
			return true
		}
	}

	context.Error(errors.NewError(nil, fmt.Sprintf("Invalid LIMIT value %v.", actual)))
	return false
}

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)))
		}
	}
}

func (this *Set) processItem(item value.AnnotatedValue, context *Context) bool {
	clone, ok := item.GetAttachment("clone").(value.AnnotatedValue)
	if !ok {
		context.Error(errors.NewError(nil,
			fmt.Sprintf("Invalid UPDATE clone of type %T.", clone)))
		return false
	}

	var e error
	for _, t := range this.plan.Node().Terms() {
		clone, e = setPath(t, clone, item, context)
		if e != nil {
			context.Error(errors.NewError(e, "Error evaluating SET clause."))
			return false
		}
	}

	item.SetAttachment("clone", clone)
	return this.sendItem(item)
}

func (this *SendUpdate) beforeItems(context *Context, parent value.Value) bool {
	if this.plan.Limit() == nil {
		return true
	}

	limit, err := this.plan.Limit().Evaluate(parent, context)
	if err != nil {
		context.Error(errors.NewError(err, ""))
		return false
	}

	switch l := limit.Actual().(type) {
	case float64:
		this.limit = int64(l)
	default:
		context.Error(errors.NewError(nil, fmt.Sprintf("Invalid LIMIT %v of type %T.", l, l)))
		return false
	}

	return true
}

func (this *spanScan) scan(context *Context, conn *datastore.IndexConnection) {
	defer context.Recover() // Recover from any panic

	dspan, err := evalSpan(this.span, context)
	if err != nil {
		context.Error(errors.NewError(err, "Error evaluating span."))
		close(conn.EntryChannel())
		return
	}

	this.plan.Index().Scan(dspan, this.plan.Distinct(), this.plan.Limit(),
		context.ScanConsistency(), context.ScanVector(), conn)
}

func (this *Filter) processItem(item value.AnnotatedValue, context *Context) bool {
	val, e := this.plan.Condition().Evaluate(item, context)
	if e != nil {
		context.Error(errors.NewError(e, "Error evaluating filter."))
		return false
	}

	if val.Truth() {
		return this.sendItem(item)
	} else {
		return true
	}
}

func (this *InitialGroup) processItem(item value.AnnotatedValue, context *Context) bool {
	// Generate the group key
	var gk string
	if len(this.plan.Keys()) > 0 {
		var e error
		gk, e = groupKey(item, this.plan.Keys(), context)
		if e != nil {
			context.Error(errors.NewError(e, "Error evaluating GROUP key."))
			return false
		}
	}

	// Get or seed the group value
	gv := this.groups[gk]
	if gv == nil {
		gv = item
		this.groups[gk] = gv

		aggregates := make(map[string]value.Value)
		gv.SetAttachment("aggregates", aggregates)
		for _, agg := range this.plan.Aggregates() {
			aggregates[agg.String()] = agg.Default()
		}
	}

	// Cumulate aggregates
	aggregates := gv.GetAttachment("aggregates").(map[string]value.Value)
	for _, agg := range this.plan.Aggregates() {
		v, e := agg.CumulateInitial(item, aggregates[agg.String()], context)
		if e != nil {
			context.Error(errors.NewError(e, "Error updating GROUP value."))
			return false
		}

		aggregates[agg.String()] = v
	}

	return true
}

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 *Unset) processItem(item value.AnnotatedValue, context *Context) bool {
	clone, ok := item.GetAttachment("clone").(value.AnnotatedValue)
	if !ok {
		context.Error(errors.NewError(nil,
			fmt.Sprintf("Invalid UPDATE clone of type %T.", clone)))
		return false
	}

	for _, t := range this.plan.Node().Terms() {
		unsetPath(t, clone, context)
	}

	return this.sendItem(item)
}

/*
Qualify identifiers for the keyspace. It also makes sure that the
keyspace term contains a name or alias.
*/
func (this *KeyspaceRef) Formalize() (f *expression.Formalizer, err error) {
	keyspace := this.Alias()
	if keyspace == "" {
		err = errors.NewError(nil, "Keyspace term must have a name or alias.")
		return
	}

	allowed := value.NewValue(make(map[string]interface{}))
	allowed.SetField(keyspace, keyspace)

	f = expression.NewFormalizer()
	f.Keyspace = keyspace
	f.Allowed = allowed
	return
}