// simplest possible implementation
// 1.  read all plans off plan channel
// 2.  return last plan
func (this *SimpleOptimizer) Optimize(planChannel plan.PlanChannel, errChannel query.ErrorChannel) (*plan.Plan, query.Error) {

	plans := make([]plan.Plan, 0)

	var p plan.Plan
	var err query.Error
	ok := true
	for ok {
		select {
		case p, ok = <-planChannel:
			if ok {
				clog.To(optimizer.CHANNEL, "See plan %v", p)
				plans = append(plans, p)
			}
		case err, ok = <-errChannel:
			if err != nil {
				return nil, err
			}
		}
	}

	if len(plans) > 0 {
		chosenPlan := plans[len(plans)-1]
		clog.To(optimizer.CHANNEL, "Choosing plan %v", chosenPlan)
		return &chosenPlan, nil
	}

	return nil, query.NewError(nil, "No plans produced for optimizer to choose from")
}

func (this *HttpQuery) Process() {
	err := this.response.Process()
	if err != nil {
		clog.To(CHANNEL, "error writing to client, aborting query")
		this.StopProcessing()
	} else {
		clog.To(CHANNEL, "response complete")
	}
}

func (this *KeyJoin) Run(stopChannel misc.StopChannel) {
	clog.To(CHANNEL, "key join operator starting")
	if this.Base.Source != nil {
		this.Base.RunOperator(this, stopChannel)
	} else {
		this.Base.SendError(query.NewError(fmt.Errorf("missing source operator"), ""))
	}

	clog.To(CHANNEL, "key join operator finished, fetched %d", this.rowsFetched)
}

func (this *StubSource) Run(stopChannel misc.StopChannel) {
	clog.To(CHANNEL, "stub source operator starting")
	defer close(this.itemChannel)
	defer close(this.supportChannel)

	for _, item := range this.data {
		this.itemChannel <- item
	}

	clog.To(CHANNEL, "stub source operator finished")
}

func (this *InterpretedExecutor) executeInternal(optimalPlan *plan.Plan, q network.Query, timeoutStopChannel misc.StopChannel) {

	clog.To(executor.CHANNEL, "simple executor started")

	// first make the plan excutable
	executablePipeline, berr := this.xpipelinebuilder.Build(optimalPlan, q)
	if berr != nil {
		q.Response().SendError(query.NewError(berr, ""))
		return
	}
	root := executablePipeline.Root

	// create a stop channel
	stopChannel := make(misc.StopChannel)
	// set it on the query object, so HTTP layer can
	// stop us if the client goes away
	q.SetStopChannel(stopChannel)
	go root.Run(stopChannel)

	// now execute it
	var item *dparval.Value
	var obj interface{}
	sourceItemChannel, supportChannel := root.GetChannels()
	ok := true
	for ok {
		select {
		case item, ok = <-sourceItemChannel:
			if ok {
				ok = this.processItem(q, item)
				clog.To(executor.CHANNEL, "simple executor sent client item: %v", item)
			}
		case obj, ok = <-supportChannel:
			if ok {
				switch obj := obj.(type) {
				case query.Error:
					q.Response().SendError(obj)
					clog.To(executor.CHANNEL, "simple executor sent client error: %v", obj)
					if obj.IsFatal() {
						return
					}
				}
			}
		case _, ok = <-timeoutStopChannel:
			clog.To(executor.CHANNEL, "simple execution aborted, timeout")
			return
		}
	}

	q.Response().NoMoreResults()
	clog.To(executor.CHANNEL, "simple executor finished")
}

func (this *EliminateDuplicates) afterItems() {
	// write the output
	for pos, item := range this.buffer {
		// we will nil out duplicates and then skip over those entries in the buffer
		if item != nil {
			if pos < len(this.buffer) {
				// look to see if the exact same item appears later in the buffer
				for nextpos, nextitem := range this.buffer[pos+1:] {
					itemProj, ok := item.GetAttachment("projection").(*dparval.Value)
					if ok {
						itemVal := itemProj.Value()
						if nextitem != nil {
							nextItemProj, ok := nextitem.GetAttachment("projection").(*dparval.Value)
							if ok {
								nextItemVal := nextItemProj.Value()
								comp := ast.CollateJSON(itemVal, nextItemVal)
								if comp == 0 {
									this.buffer[pos+nextpos+1] = nil
								}
							}
						}
					}
				}
			}
			clog.To(DEBUG_DUP_CHANNEL, "distinct: %v", item)
			this.Base.SendItem(item)
		}
	}
}

func newPool(s *site, name string) (*pool, query.Error) {
	clog.To(catalog.CHANNEL, "Created New Pool %s", name)
	cbpool, err := s.client.GetPool(name)
	if err != nil {
		if name == "default" {
			// if default pool is not available, try reconnecting to the server
			url := s.URL()
			client, err := cb.Connect(url)
			if err != nil {
				return nil, query.NewError(nil, fmt.Sprintf("Pool %v not found.", name))
			}
			// check if the default pool exists
			cbpool, err = client.GetPool(name)
			if err != nil {
				return nil, query.NewError(nil, fmt.Sprintf("Pool %v not found.", name))
			}
			s.client = client
		}
	}
	rv := pool{
		site:        s,
		name:        name,
		cbpool:      cbpool,
		bucketCache: make(map[string]catalog.Bucket),
	}
	go keepPoolFresh(&rv)
	return &rv, nil
}

func newBucket(p *pool, name string) (*bucket, query.Error) {
	clog.To(catalog.CHANNEL, "Created New Bucket %s", name)
	cbbucket, err := p.cbpool.GetBucket(name)
	if err != nil {
		// go-couchbase caches the buckets
		// to be sure no such bucket exists right now
		// we trigger a refresh
		p.refresh()
		// and then check one more time
		cbbucket, err = p.cbpool.GetBucket(name)
		if err != nil {
			// really no such bucket exists
			return nil, query.NewError(nil, fmt.Sprintf("Bucket %v not found.", name))
		}
	}

	rv := &bucket{
		pool:     p,
		name:     name,
		cbbucket: cbbucket,
		indexes:  make(map[string]catalog.Index),
	}

	ierr := rv.loadIndexes()
	if err != nil {
		return nil, ierr
	}

	return rv, nil
}

func (this *HttpEndpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	clog.To(CHANNEL, "request received")
	q := NewHttpQuery(w, r)
	if q != nil {
		this.queryChannel <- q
		q.Process()
	}
}

func (this *Fetch) Run(stopChannel misc.StopChannel) {
	clog.To(CHANNEL, "fetch operator starting")
	if this.Base.Source != nil {
		this.Base.RunOperator(this, stopChannel)
	} else {
		defer close(this.Base.itemChannel)
		defer close(this.Base.supportChannel)
		defer close(this.Base.upstreamStopChannel)
		for _, id := range this.ids {
			doc := dparval.NewValue(map[string]interface{}{})
			doc.SetAttachment("meta", map[string]interface{}{"id": id})
			this.processItem(doc)
		}
		this.afterItems()
	}
	clog.To(CHANNEL, "fetch operator finished, fetched %d", this.rowsFetched)
}

func (this *KeyScan) Run(stopChannel misc.StopChannel) {
	defer close(this.itemChannel)
	defer close(this.supportChannel)
	// this MUST be here so that it runs before the channels are closed
	defer this.RecoverPanic()

	clog.To(CHANNEL, "key scan operator starting")

	for _, item := range this.keylist {
		this.rowsKeyScanned += 1
		// rematerialize an object from the data returned by this index entry
		doc := dparval.NewValue(map[string]interface{}{})
		// attach metadata
		doc.SetAttachment("meta", map[string]interface{}{"id": item})
		this.SendItem(doc)
	}

	clog.To(CHANNEL, "key scan operator finished, scanned %d", this.rowsKeyScanned)
}

func (p *pool) refresh() {
	// trigger refresh of this pool
	clog.To(catalog.CHANNEL, "Refreshing Pool %s", p.name)
	newpool, err := p.site.client.GetPool(p.name)
	if err != nil {
		clog.Warnf("Error updating pool: %v", err)
		return
	}
	p.cbpool = newpool
}

func WalkViewInBatches(result chan cb.ViewRow, errs query.ErrorChannel, bucket *cb.Bucket,
	ddoc string, view string, options map[string]interface{}, batchSize int64, limit int64) {

	if limit != 0 && limit < batchSize {
		batchSize = limit
	}

	defer close(result)
	defer close(errs)

	defer func() {
		r := recover()
		if r != nil {
			clog.Error(fmt.Errorf("View Walking Panic: %v\n%s", r, debug.Stack()))
			errs <- query.NewError(nil, "Panic In View Walking")
		}
	}()

	options["limit"] = batchSize + 1

	numRead := int64(0)
	ok := true
	for ok {

		logURL, err := bucket.ViewURL(ddoc, view, options)
		if err == nil {
			clog.To(NETWORK_CHANNEL, "Request View: %v", logURL)
		}
		vres, err := bucket.View(ddoc, view, options)
		if err != nil {
			errs <- query.NewError(err, "Unable to access view")
			return
		}

		for i, row := range vres.Rows {
			if int64(i) < batchSize {
				// dont process the last row, its just used to see if we
				// need to continue processing
				result <- row
				numRead += 1
			}
		}

		if (int64(len(vres.Rows)) > batchSize) && (limit == 0 || (limit != 0 && numRead < limit)) {
			// prepare for next run
			skey := vres.Rows[batchSize].Key
			skeydocid := vres.Rows[batchSize].ID
			options["startkey"] = skey
			options["startkey_docid"] = cb.DocID(skeydocid)
		} else {
			// stop
			ok = false
		}
	}
}

func (this *DropIndex) Run(stopChannel misc.StopChannel) {
	defer close(this.itemChannel)
	defer close(this.supportChannel)
	// this MUST be here so that it runs before the channels are closed
	defer this.RecoverPanic()

	this.downstreamStopChannel = stopChannel
	clog.To(CHANNEL, "drop_index operator starting")
	err := this.index.Drop()
	if err != nil {
		this.SendError(err)
	} else {
		item := dparval.NewValue(map[string]interface{}{})
		item.SetAttachment("projection", map[string]interface{}{
			"dropped": true,
		})
		this.SendItem(item)
	}
	clog.To(CHANNEL, "drop_index operator finished")
}

func (this *CreateIndex) Run(stopChannel misc.StopChannel) {
	defer close(this.itemChannel)
	defer close(this.supportChannel)
	// this MUST be here so that it runs before the channels are closed
	defer this.RecoverPanic()

	indexType := catalog.IndexType(strings.ToLower(this.index_type))
	indexOn := make(catalog.IndexKey, len(this.on))
	for pos, key := range this.on {
		indexOn[pos] = key
	}

	this.downstreamStopChannel = stopChannel

	var index catalog.Index
	var err query.Error

	if this.primary {
		clog.To(CHANNEL, "create_index (primary) operator starting")
		index, err = this.bucket.CreatePrimaryIndex()
	} else {
		clog.To(CHANNEL, "create_index (secondary) operator starting")
		index, err = this.bucket.CreateIndex(this.name, indexOn, indexType)
	}

	if err != nil {
		this.SendError(err)
	} else {
		if index != nil {
			item := dparval.NewValue(map[string]interface{}{})
			item.SetAttachment("projection", map[string]interface{}{
				"id":   index.Id(),
				"name": index.Name(),
			})
			this.SendItem(item)
		} else {
			clog.Warn("Successfully created index, but index was nil")
		}
	}
	clog.To(CHANNEL, "create_index operator finished")
}

func (this *Explain) Run(stopChannel misc.StopChannel) {
	defer close(this.itemChannel)
	defer close(this.supportChannel)
	// this MUST be here so that it runs before the channels are closed
	defer this.RecoverPanic()

	this.downstreamStopChannel = stopChannel
	clog.To(CHANNEL, "explain operator starting")
	item := dparval.NewValue(map[string]interface{}{})

	planBytes, err := json.Marshal(this.Plan)
	if err != nil {
		this.SendError(query.NewError(err, "error serializing plan to JSON"))
	} else {
		projection := dparval.NewValueFromBytes(planBytes)
		item.SetAttachment("projection", projection)
		this.SendItem(item)
	}

	clog.To(CHANNEL, "explain operator finished")
}

func CanIUseThisIndexForThisWhereClause(index catalog.RangeIndex, where ast.Expression, bucket string) (bool, plan.ScanRanges, ast.Expression, error) {

	// convert the index key to formal notation
	indexKeyFormal, err := IndexKeyInFormalNotation(index.Key(), bucket)
	if err != nil {
		return false, nil, nil, err
	}

	// put the where clause into conjunctive normal form
	ennf := ast.NewExpressionNNF()
	whereNNF, err := where.Accept(ennf)
	if err != nil {
		return false, nil, nil, err
	}
	ecnf := ast.NewExpressionCNF()
	whereCNF, err := whereNNF.Accept(ecnf)
	if err != nil {
		return false, nil, nil, err
	}

	switch whereCNF := whereCNF.(type) {
	case *ast.AndOperator:
		// this is an and, we can try to satisfy individual operands
		found := false
		rranges := plan.ScanRanges{}
		for _, oper := range whereCNF.Operands {
			// see if the where clause expression is sargable with respect to the index key
			es := NewExpressionSargable(indexKeyFormal[0])
			oper.Accept(es)
			if es.IsSargable() {
				found = true
				for _, ran := range es.ScanRanges() {
					rranges = MergeRanges(rranges, ran)
					clog.To(planner.CHANNEL, "now ranges are: %v", rranges)
				}
			}
		}
		if found {
			return true, rranges, nil, nil
		}
	default:
		// not an and, we must satisfy the whole expression
		// see if the where clause expression is sargable with respect to the index key
		es := NewExpressionSargable(indexKeyFormal[0])
		whereCNF.Accept(es)
		if es.IsSargable() {
			return true, es.ScanRanges(), nil, nil
		}
	}

	// cannot use this index
	return false, nil, nil, nil
}

// NewSite creates a new Couchbase site for the given url.
func NewSite(url string) (catalog.Site, query.Error) {
	clog.To(catalog.CHANNEL, "Created New Site %s", url)
	client, err := cb.Connect(url)

	if err != nil {
		return nil, query.NewError(err, "")
	}

	return &site{
		client:    client,
		poolCache: make(map[string]catalog.Pool),
	}, nil
}

func (this *InterpretedExecutor) Execute(optimalPlan *plan.Plan, q network.Query, timeout *time.Duration) {
	stopChannel := make(misc.StopChannel)
	if timeout.Nanoseconds() < 0 {
		this.executeInternal(optimalPlan, q, stopChannel)
	} else {
		c := make(chan error, 1)
		go func() {
			this.executeInternal(optimalPlan, q, stopChannel)
			c <- nil
		}()
		select {
		case <-c:
			return
		case <-time.After(*timeout):
			clog.To(executor.CHANNEL, "simple executor timeout trigger")
			close(stopChannel)
			clog.To(executor.CHANNEL, "stop channel closed")
		}
		<-c
		q.Response().SendError(query.NewTimeoutError(timeout))
	}
}

func ViewTotalRows(bucket *cb.Bucket, ddoc string, view string, options map[string]interface{}) (int64, query.Error) {
	options["limit"] = 0

	logURL, err := bucket.ViewURL(ddoc, view, options)
	if err == nil {
		clog.To(NETWORK_CHANNEL, "Request View: %v", logURL)
	}
	vres, err := bucket.View(ddoc, view, options)
	if err != nil {
		return 0, query.NewError(err, "Unable to access view")
	}

	return int64(vres.TotalRows), nil
}