// Decr increments stat value(s) by `vals`.
func (s Statistics) Decr(path string, vals ...int) {
	l := len(vals)
	if l == 0 {
		logging.Warnf("Decr called without value")
		return
	}

	switch vs := s[path].(type) {
	case float64:
		s[path] = vs - float64(vals[0])

	case []interface{}:
		if l != len(vs) {
			logging.Warnf("Decr expected %v values, got %v", len(vs), l)
			return
		}
		for i, v := range vs {
			vs[i] = v.(float64) - float64(vals[i])
		}

	case []float64:
		if l != len(vs) {
			logging.Warnf("Incr expected %v values, got %v", len(vs), l)
			return
		}
		for i, v := range vs {
			vs[i] = v - float64(vals[i])
		}
	}
}

func makeTapEvent(req transport.MCRequest) *TapEvent {
	event := TapEvent{
		VBucket: req.VBucket,
	}
	switch req.Opcode {
	case transport.TAP_MUTATION:
		event.Opcode = TapMutation
		event.Key = req.Key
		event.Value = req.Body
		event.Cas = req.Cas
	case transport.TAP_DELETE:
		event.Opcode = TapDeletion
		event.Key = req.Key
		event.Cas = req.Cas
	case transport.TAP_CHECKPOINT_START:
		event.Opcode = TapCheckpointStart
	case transport.TAP_CHECKPOINT_END:
		event.Opcode = TapCheckpointEnd
	case transport.TAP_OPAQUE:
		if len(req.Extras) < 8+4 {
			return nil
		}
		switch op := int(binary.BigEndian.Uint32(req.Extras[8:])); op {
		case transport.TAP_OPAQUE_INITIAL_VBUCKET_STREAM:
			event.Opcode = TapBeginBackfill
		case transport.TAP_OPAQUE_CLOSE_BACKFILL:
			event.Opcode = TapEndBackfill
		case transport.TAP_OPAQUE_CLOSE_TAP_STREAM:
			event.Opcode = tapEndStream
		case transport.TAP_OPAQUE_ENABLE_AUTO_NACK:
			return nil
		case transport.TAP_OPAQUE_ENABLE_CHECKPOINT_SYNC:
			return nil
		default:
			logging.Warnf("TapFeed: Ignoring TAP_OPAQUE/%d", op)
			return nil // unknown opaque event
		}
	case transport.NOOP:
		return nil // ignore
	default:
		logging.Warnf("TapFeed: Ignoring %s", req.Opcode)
		return nil // unknown event
	}

	if len(req.Extras) >= tapMutationExtraLen &&
		(event.Opcode == TapMutation || event.Opcode == TapDeletion) {

		event.Flags = binary.BigEndian.Uint32(req.Extras[8:])
		event.Expiry = binary.BigEndian.Uint32(req.Extras[12:])
	}

	return &event
}

//
// Run the server until it stop.  Will not attempt to re-run.
//
func (c *Coordinator) runOnce(config string) int {

	logging.Debugf("Coordinator.runOnce() : Start Running Coordinator")

	pauseTime := 0

	defer func() {
		if r := recover(); r != nil {
			logging.Warnf("panic in Coordinator.runOnce() : %s\n", r)
		}

		common.SafeRun("Coordinator.cleanupState()",
			func() {
				c.cleanupState()
			})
	}()

	err := c.bootstrap(config)
	if err != nil {
		pauseTime = 200
	}

	// Check if the server has been terminated explicitly. If so, don't run.
	if !c.IsDone() {

		// runElection() finishes if there is an error, election result is known or
		// it being terminated. Unless being killed explicitly, a goroutine
		// will continue to run to responds to other peer election request
		leader, err := c.runElection()
		if err != nil {
			logging.Warnf("Coordinator.runOnce() : Error Encountered During Election : %s", err.Error())
			pauseTime = 100
		} else {

			// Check if the server has been terminated explicitly. If so, don't run.
			if !c.IsDone() {
				// runCoordinator() is done if there is an error	or being terminated explicitly (killch)
				err := c.runProtocol(leader)
				if err != nil {
					logging.Warnf("Coordinator.RunOnce() : Error Encountered From Coordinator : %s", err.Error())
				}
			}
		}
	} else {
		logging.Infof("Coordinator.RunOnce(): Coordinator has been terminated explicitly. Terminate.")
	}

	return pauseTime
}

// Goroutine that runs the feed
func (feed *TapFeed) run() {
	retryInterval := initialRetryInterval
	bucketOK := true
	for {
		// Connect to the TAP feed of each server node:
		if bucketOK {
			killSwitch, err := feed.connectToNodes()
			if err == nil {
				// Run until one of the sub-feeds fails:
				select {
				case <-killSwitch:
				case <-feed.quit:
					return
				}
				feed.closeNodeFeeds()
				retryInterval = initialRetryInterval
			}
		}

		// On error, try to refresh the bucket in case the list of nodes changed:
		logging.Warnf("dcp-client: TAP connection lost; reconnecting to bucket %q in %v",
			feed.bucket.Name, retryInterval)
		err := feed.bucket.Refresh()
		bucketOK = err == nil

		select {
		case <-time.After(retryInterval):
		case <-feed.quit:
			return
		}
		if retryInterval *= 2; retryInterval > maximumRetryInterval {
			retryInterval = maximumRetryInterval
		}
	}
}

//
// Terminate the Coordinator
//
func (s *Coordinator) Terminate() {

	defer func() {
		if r := recover(); r != nil {
			logging.Warnf("panic in Coordinator.Terminate() : %s.  Ignored.\n", r)
		}
	}()

	s.state.mutex.Lock()
	defer s.state.mutex.Unlock()

	if s.state.done {
		return
	}

	s.state.done = true

	if s.site != nil {
		s.site.Close()
		s.site = nil
	}

	if s.configRepo != nil {
		s.configRepo.Close()
		s.configRepo = nil
	}

	if s.skillch != nil {
		s.skillch <- true // kill leader/follower server
	}
}

func (s *Server) monitorClient(
	conn net.Conn,
	rcvch <-chan interface{},
	quitch chan<- interface{},
	finch chan bool) {

	raddr := conn.RemoteAddr()

	select {
	case req, ok := <-rcvch:
		if ok {
			if _, yes := req.(*protobuf.EndStreamRequest); yes {
				format := "%v connection %s client requested quit"
				logging.Debugf(format, s.logPrefix, raddr)
			} else {
				format := "%v connection %s unknown request %v"
				logging.Errorf(format, s.logPrefix, raddr, req)
			}
		} else {
			format := "%v connection %s client closed connection"
			logging.Warnf(format, s.logPrefix, raddr)
		}
	case <-s.killch:
	case <-finch:
		close(finch)
		return
	}
	close(quitch)

	<-finch
	close(finch)
}

func (p *Pool) refresh() (err error) {
	p.BucketMap = make(map[string]Bucket)

loop:
	buckets := []Bucket{}
	err = p.client.parseURLResponse(p.BucketURL["uri"], &buckets)
	if err != nil {
		return err
	}
	for _, b := range buckets {
		nb := &Bucket{}
		err = p.client.parseURLResponse(p.BucketURL["terseBucketsBase"]+b.Name, nb)
		if err != nil {
			// bucket list is out of sync with cluster bucket list
			// bucket might have got deleted.
			if strings.Contains(err.Error(), "HTTP error 404") {
				logging.Warnf("cluster_info: Out of sync for bucket %s. Retrying..", b.Name)
				goto loop
			}
			return err
		}
		b.pool = p
		b.init(nb)
		p.BucketMap[b.Name] = b
	}
	return nil
}

func (instance *serviceNotifierInstance) getNotifyCallback(t NotificationType) func(interface{}) error {
	fn := func(msg interface{}) error {
		instance.Lock()
		defer instance.Unlock()

		if !instance.valid {
			return ErrNotifierInvalid
		}

		notifMsg := Notification{
			Type: t,
			Msg:  msg,
		}

		logging.Infof("serviceChangeNotifier: received %s", notifMsg)

		for id, w := range instance.waiters {
			select {
			case w <- notifMsg:
			case <-time.After(notifyWaitTimeout):
				logging.Warnf("servicesChangeNotifier: Consumer for %v took too long to read notification, making the consumer invalid", instance.clusterUrl)
				close(w)
				delete(instance.waiters, id)
			}
		}
		return nil
	}

	return fn
}

// Update config object with data, can be a Config, map[string]interface{},
// []byte.
func (config Config) Update(data interface{}) error {
	fmsg := "CONF[] skipping setting key %q value '%v': %v"
	switch v := data.(type) {
	case Config: // Clone
		for key, value := range v {
			config.Set(key, value)
		}

	case []byte: // parse JSON
		m := make(map[string]interface{})
		if err := json.Unmarshal(v, &m); err != nil {
			return err
		}
		config.Update(m)

	case map[string]interface{}: // transform
		for key, value := range v {
			if cv, ok := SystemConfig[key]; ok { // valid config.
				if _, ok := config[key]; !ok {
					config[key] = cv // copy by value
				}
				if err := config.SetValue(key, value); err != nil {
					logging.Warnf(fmsg, key, value, err)
				}

			} else {
				logging.Errorf("invalid config param %q", key)
			}
		}

	default:
		return nil
	}
	return nil
}

func waitForConnections(ls net.Listener) {
	reqChannel := make(chan chanReq)

	go RunServer(reqChannel)
	handler := &reqHandler{reqChannel}

	logging.Warnf("Listening on port %d", *port)
	for {
		s, e := ls.Accept()
		if e == nil {
			logging.Warnf("Got a connection from %v", s.RemoteAddr())
			go connectionHandler(s, handler)
		} else {
			logging.Warnf("Error accepting from %s", ls)
		}
	}
}

//GetPartitionById returns the partition for the given partitionId
//or nil if partitionId is not found
func (pc *KeyPartitionContainer) GetPartitionById(id PartitionId) PartitionDefn {
	if p, ok := pc.PartitionMap[id]; ok {
		return p
	} else {
		logging.Warnf("KeyPartitionContainer: Invalid Partition Id %v", id)
		return nil
	}
}

func (instance *serviceNotifierInstance) RunServicesObserver() {
	servicesCallback := instance.getNotifyCallback(ServiceChangeNotification)
	err := instance.client.RunObserveNodeServices(instance.pool, servicesCallback, nil)
	if err != nil {
		logging.Warnf("servicesChangeNotifier: Connection terminated for services notifier instance of %s, %s (%v)", instance.clusterUrl, instance.pool, err)
	}
	instance.cleanup()
}

// Internal goroutine that reads from the socket and writes events to
// the channel
func (mc *Client) runFeed(ch chan TapEvent, feed *TapFeed) {
	defer close(ch)
	var headerBuf [transport.HDR_LEN]byte
loop:
	for {
		// Read the next request from the server.
		//
		//  (Can't call mc.Receive() because it reads a
		//  _response_ not a request.)
		var pkt transport.MCRequest
		n, err := pkt.Receive(mc.conn, headerBuf[:])
		if TapRecvHook != nil {
			TapRecvHook(&pkt, n, err)
		}

		if err != nil {
			if err != io.EOF {
				feed.Error = err
			}
			break loop
		}

		//logging.Warnf("** TapFeed received %#v : %q", pkt, pkt.Body)

		if pkt.Opcode == transport.TAP_CONNECT {
			// This is not an event from the server; it's
			// an error response to my connect request.
			feed.Error = fmt.Errorf("tap connection failed: %s", pkt.Body)
			break loop
		}

		event := makeTapEvent(pkt)
		if event != nil {
			if event.Opcode == tapEndStream {
				break loop
			}

			select {
			case ch <- *event:
			case <-feed.closer:
				break loop
			}
		}

		if len(pkt.Extras) >= 4 {
			reqFlags := binary.BigEndian.Uint16(pkt.Extras[2:])
			if reqFlags&transport.TAP_ACK != 0 {
				if _, err := mc.sendAck(&pkt); err != nil {
					feed.Error = err
					break loop
				}
			}
		}
	}
	if err := mc.Close(); err != nil {
		logging.Warnf("Error closing memcached client:  %v", err)
	}
}

func handleFlush(req *transport.MCRequest, s *storage) (ret *transport.MCResponse) {
	ret = &transport.MCResponse{}
	delay := binary.BigEndian.Uint32(req.Extras)
	if delay > 0 {
		logging.Warnf("Delay not supported (got %d)", delay)
	}
	s.data = make(map[string]transport.MCItem)
	return
}

// RunServer runs the cache server.
func RunServer(input chan chanReq) {
	var s storage
	s.data = make(map[string]transport.MCItem)
	for {
		req := <-input
		logging.Warnf("Got a request: %s", req.req)
		req.res <- dispatch(req.req, &s)
	}
}

//GetEndpointsByPartitionId returns the list of Endpoints hosting the give partitionId
//or nil if partitionId is not found
func (pc *KeyPartitionContainer) GetEndpointsByPartitionId(id PartitionId) []Endpoint {

	if p, ok := pc.PartitionMap[id]; ok {
		return p.Endpoints()
	} else {
		logging.Warnf("KeyPartitionContainer: Invalid Partition Id %v", id)
		return nil
	}
}

//GetSliceById returns Slice for the given SliceId
func (sc *HashedSliceContainer) GetSliceById(id SliceId) Slice {

	if s, ok := sc.SliceMap[id]; ok {
		return s
	} else {
		logging.Warnf("HashedSliceContainer: Invalid Slice Id %v", id)
		return nil
	}
}

func (p *Pipeline) runIt(o *pipelineObject) {
	p.wg.Add(1)
	go func() {
		err := o.r.Routine()
		if err != nil {
			logging.Warnf("%v exited with error %v", o.n, err)
		}
		p.wg.Done()
	}()
}

// LogConfig will check wether a configuration parameter is
// mutable and log that information.
func (config Config) LogConfig(prefix string) {
	for key, cv := range config {
		if cv.Immutable {
			fmsg := "%v immutable settings %v cannot be update to `%v`\n"
			logging.Warnf(fmsg, prefix, key, cv.Value)
		} else {
			fmsg := "%v settings %v will updated to `%v`\n"
			logging.Infof(fmsg, prefix, key, cv.Value)
		}
	}
}

func (c *clustMgrAgent) handleGetGlobalTopology(cmd Message) {

	logging.Debugf("ClustMgr:handleGetGlobalTopology %v", cmd)

	//get the latest topology from manager
	metaIter, err := c.mgr.NewIndexDefnIterator()
	if err != nil {
		common.CrashOnError(err)
	}
	defer metaIter.Close()

	indexInstMap := make(common.IndexInstMap)

	for _, defn, err := metaIter.Next(); err == nil; _, defn, err = metaIter.Next() {

		var idxDefn common.IndexDefn
		idxDefn = *defn

		t, e := c.mgr.GetTopologyByBucket(idxDefn.Bucket)
		if e != nil {
			common.CrashOnError(e)
		}

		inst := t.GetIndexInstByDefn(idxDefn.DefnId)

		if inst == nil {
			logging.Warnf("ClustMgr:handleGetGlobalTopology Index Instance Not "+
				"Found For Index Definition %v. Ignored.", idxDefn)
			continue
		}

		//for indexer, Ready state doesn't matter. Till index build,
		//the index stays in Created state.
		var state common.IndexState
		instState := common.IndexState(inst.State)
		if instState == common.INDEX_STATE_READY {
			state = common.INDEX_STATE_CREATED
		} else {
			state = instState
		}

		idxInst := common.IndexInst{InstId: common.IndexInstId(inst.InstId),
			Defn:   idxDefn,
			State:  state,
			Stream: common.StreamId(inst.StreamId),
		}

		indexInstMap[idxInst.InstId] = idxInst

	}

	c.supvCmdch <- &MsgClustMgrTopology{indexInstMap: indexInstMap}
}