func indexDefnId(key string) (common.IndexDefnId, error) {
	val, err := strconv.ParseUint(key, 10, 64)
	if err != nil {
		return common.IndexDefnId(0), err
	}
	return common.IndexDefnId(val), nil
}

func (m *LifecycleMgr) UpdateIndexInstance(bucket string, defnId common.IndexDefnId, state common.IndexState,
	streamId common.StreamId, errStr string, buildTime []uint64) error {

	topology, err := m.repo.GetTopologyByBucket(bucket)
	if err != nil {
		logging.Errorf("LifecycleMgr.handleTopologyChange() : index instance update fails. Reason = %v", err)
		return err
	}

	changed := false
	if state != common.INDEX_STATE_NIL {
		changed = topology.UpdateStateForIndexInstByDefn(common.IndexDefnId(defnId), common.IndexState(state)) || changed
	}

	if streamId != common.NIL_STREAM {
		changed = topology.UpdateStreamForIndexInstByDefn(common.IndexDefnId(defnId), common.StreamId(streamId)) || changed
	}

	changed = topology.SetErrorForIndexInstByDefn(common.IndexDefnId(defnId), errStr) || changed

	if changed {
		if err := m.repo.SetTopologyByBucket(bucket, topology); err != nil {
			logging.Errorf("LifecycleMgr.handleTopologyChange() : index instance update fails. Reason = %v", err)
			return err
		}
	}

	return nil
}

// start up
func changeTopologyForTimerTest(mgr *manager.IndexManager) {

	// Add a new index definition : 406
	idxDefn := &common.IndexDefn{
		DefnId:          common.IndexDefnId(406),
		Name:            "stream_mgr_timer_test",
		Using:           common.ForestDB,
		Bucket:          "Default",
		IsPrimary:       false,
		SecExprs:        []string{"Testing"},
		ExprType:        common.N1QL,
		PartitionScheme: common.HASH,
		PartitionKey:    "Testing"}

	logging.Infof("Run Timer Test : Create Index Defn 406")
	if err := mgr.HandleCreateIndexDDL(idxDefn); err != nil {
		util.TT.Fatal(err)
	}
	// Wait so there is no race condition.
	time.Sleep(time.Duration(1000) * time.Millisecond)

	// Update the index definition to ready
	logging.Infof("Run Timer Test : Update Index Defn 406 to READY")
	topology, err := mgr.GetTopologyByBucket("Default")
	if err != nil {
		util.TT.Fatal(err)
	}
	topology.ChangeStateForIndexInstByDefn(common.IndexDefnId(406), common.INDEX_STATE_CREATED, common.INDEX_STATE_READY)
	if err := mgr.SetTopologyByBucket("Default", topology); err != nil {
		util.TT.Fatal(err)
	}
}

func extractDefnIdFromKey(key string) (c.IndexDefnId, error) {
	i := strings.Index(key, "/")
	if i != -1 && i < len(key)-1 {
		id, err := strconv.ParseUint(key[i+1:], 10, 64)
		return c.IndexDefnId(id), err
	}

	return c.IndexDefnId(0), errors.New("watcher.processChange() : cannot parse index definition id")
}

func (m *LifecycleMgr) handleDeleteBucket(bucket string, content []byte) error {

	result := error(nil)

	if len(content) == 0 {
		return errors.New("invalid argument")
	}

	streamId := common.StreamId(content[0])

	topology, err := m.repo.GetTopologyByBucket(bucket)
	if err == nil {
		/*
			// if there is an error getting the UUID, this means that
			// the node is not able to connect to pool service in order
			// to fetch the bucket UUID.   Return an error and skip.
			uuid, err := m.getBucketUUID(bucket)
			if err != nil {
				logging.Errorf("LifecycleMgr.handleDeleteBucket() : Encounter when connecting to pool service = %v", err)
				return err
			}
		*/

		// At this point, we are able to connect to pool service.  If pool
		// does not contain the bucket, then we delete all index defn in
		// the bucket.  Otherwise, delete index defn that does not have the
		// same bucket UUID.  Note that any other create index request will
		// be blocked while this call is run.
		definitions := make([]IndexDefnDistribution, len(topology.Definitions))
		copy(definitions, topology.Definitions)

		for _, defnRef := range definitions {

			if defn, err := m.repo.GetIndexDefnById(common.IndexDefnId(defnRef.DefnId)); err == nil {

				logging.Debugf("LifecycleMgr.handleDeleteBucket() : index instance: id %v, streamId %v.",
					defn.DefnId, defnRef.Instances[0].StreamId)

				// delete index defn from the bucket if bucket uuid is not specified or
				// index does *not* belong to bucket uuid
				if /* (uuid == common.BUCKET_UUID_NIL || defn.BucketUUID != uuid) && */
				streamId == common.NIL_STREAM || common.StreamId(defnRef.Instances[0].StreamId) == streamId {
					if err := m.DeleteIndex(common.IndexDefnId(defn.DefnId), false); err != nil {
						result = err
					}
				}
			} else {
				logging.Debugf("LifecycleMgr.handleDeleteBucket() : Cannot find index instance %v.  Skip.", defnRef.DefnId)
			}
		}
	} else if err != fdb.RESULT_KEY_NOT_FOUND {
		result = err
	}

	return result
}

func (b *metadataClient) deleteIndex(defnID uint64) {
	b.rw.Lock()
	defer b.rw.Unlock()

	id := common.IndexDefnId(defnID)
	for indexerID, indexes := range b.topology {
		delete(indexes, id)
		b.topology[indexerID] = indexes
	}
	b.replicas = b.computeReplicas()
	delete(b.loads, common.IndexDefnId(defnID))
}

func (w *watcher) processChange(op uint32, key string, content []byte) error {

	logging.Debugf("watcher.processChange(): key = %v", key)
	defer logging.Debugf("watcher.processChange(): done -> key = %v", key)

	opCode := common.OpCode(op)

	switch opCode {
	case common.OPCODE_ADD, common.OPCODE_SET:
		if isIndexDefnKey(key) {
			if len(content) == 0 {
				logging.Debugf("watcher.processChange(): content of key = %v is empty.", key)
			}

			id, err := extractDefnIdFromKey(key)
			if err != nil {
				return err
			}
			w.addDefnWithNoLock(c.IndexDefnId(id))
			if err := w.provider.repo.unmarshallAndAddDefn(content); err != nil {
				return err
			}
			w.notifyEventNoLock()

		} else if isIndexTopologyKey(key) {
			if len(content) == 0 {
				logging.Debugf("watcher.processChange(): content of key = %v is empty.", key)
			}
			if err := w.provider.repo.unmarshallAndAddInst(content); err != nil {
				return err
			}
			w.notifyEventNoLock()
		}
	case common.OPCODE_DELETE:
		if isIndexDefnKey(key) {

			id, err := extractDefnIdFromKey(key)
			if err != nil {
				return err
			}
			w.removeDefnWithNoLock(c.IndexDefnId(id))
			w.provider.repo.removeDefn(c.IndexDefnId(id))
			w.notifyEventNoLock()
		}
	}

	return nil
}

// IndexState implement BridgeAccessor{} interface.
func (b *metadataClient) IndexState(defnID uint64) (common.IndexState, error) {
	b.Refresh()

	b.rw.RLock()
	defer b.rw.RUnlock()

	for _, indexes := range b.topology {
		for _, index := range indexes {
			if index.Definition.DefnId == common.IndexDefnId(defnID) {
				if index.Instances != nil && len(index.Instances) > 0 {
					state := index.Instances[0].State
					if len(index.Instances) == 0 {
						err := fmt.Errorf("no instance for %q", defnID)
						return state, err
					} else if index.Instances[0].Error != "" {
						return state, errors.New(index.Instances[0].Error)
					} else {
						return state, nil
					}
				}
				return common.INDEX_STATE_ERROR, ErrorInstanceNotFound
			}
		}
	}
	return common.INDEX_STATE_ERROR, ErrorIndexNotFound
}

// DropIndex implements BridgeAccessor{} interface.
func (b *metadataClient) DropIndex(defnID uint64) error {
	err := b.mdClient.DropIndex(common.IndexDefnId(defnID))
	if err == nil { // cleanup index local cache.
		b.deleteIndex(defnID)
	}
	return err
}

// Create a mock index that uses a feeder function to provide query results.
// Creates an index with single partition, single slice with a snapshot.
func (s *scannerTestHarness) createIndex(name, bucket string, feeder snapshotFeeder) c.IndexDefnId {
	s.indexCount++

	pc := c.NewKeyPartitionContainer()
	pId := c.PartitionId(0)
	endpt := c.Endpoint("localhost:1000")
	pDef := c.KeyPartitionDefn{Id: pId, Endpts: []c.Endpoint{endpt}}
	pc.AddPartition(pId, pDef)

	instId := c.IndexInstId(s.indexCount)
	defnId := c.IndexDefnId(0xABBA)
	indDefn := c.IndexDefn{Name: name, Bucket: bucket, DefnId: defnId}
	indInst := c.IndexInst{InstId: instId, State: c.INDEX_STATE_ACTIVE,
		Defn: indDefn, Pc: pc,
	}
	// TODO: Use cmdch to update map
	s.scanner.indexInstMap[instId] = indInst

	sc := NewHashedSliceContainer()
	partInst := PartitionInst{Defn: pDef, Sc: sc}
	partInstMap := PartitionInstMap{pId: partInst}

	snap := &mockSnapshot{feeder: feeder}
	snap.SetTimestamp(s.scanTS)

	slice := &mockSlice{}
	slId := SliceId(0)
	sc.AddSlice(slId, slice)
	// TODO: Use cmdch to update map
	s.scanner.indexPartnMap[instId] = partInstMap
	return defnId
}

func (m *LifecycleMgr) verifyBucket(bucket string) (string, error) {

	currentUUID, err := m.getBucketUUID(bucket)
	if err != nil {
		return common.BUCKET_UUID_NIL, err
	}

	topology, err := m.repo.GetTopologyByBucket(bucket)
	if err != nil && err != fdb.RESULT_KEY_NOT_FOUND {
		return common.BUCKET_UUID_NIL, err
	}

	if topology != nil {
		for _, defnRef := range topology.Definitions {
			if defn, err := m.repo.GetIndexDefnById(common.IndexDefnId(defnRef.DefnId)); err == nil {
				if defn.BucketUUID != currentUUID {
					return common.BUCKET_UUID_NIL,
						errors.New("Bucket does not exist or temporarily unavaible for creating new index." +
							" Please retry the operation at a later time.")
				}
			}
		}
	}

	// topology is either nil or all index defn matches bucket UUID
	return currentUUID, nil
}

func dropIndexRequest(t *testing.T) {

	logging.Infof("********** Start dropIndexRequest")

	// Construct request body.
	info := common.IndexDefn{
		DefnId: common.IndexDefnId(500),
		Name:   "request_handler_test",
		Bucket: "Default",
	}

	req := manager.IndexRequest{Version: uint64(1), Type: manager.DROP, Index: info}
	body, err := json.Marshal(req)
	if err != nil {
		t.Fatal(err)
	}

	bodybuf := bytes.NewBuffer(body)
	resp, err := http.Post("http://localhost:9102/dropIndex", "application/json", bodybuf)
	if err != nil {
		t.Fatal(err)
	}

	validateIndexResponse(resp, t)

	logging.Infof("********** Done dropIndexRequest")
}

// getNode hosting index with `defnID`.
func (b *metadataClient) getNode(defnID uint64) (adminport string, ok bool) {
	aport, _, err := b.mdClient.FindServiceForIndex(common.IndexDefnId(defnID))
	if err != nil {
		return "", false
	}
	return aport, true
}

//
// Serialize topology into a protobuf message format
//
func convertTopologyToIndexInstProtoMsg(mgr *IndexManager,
	topology *IndexTopology, port string) ([]*protobuf.Instance, error) {

	var result []*protobuf.Instance = nil

	for _, defnRef := range topology.Definitions {

		// look up the index definition from dictionary
		defn, err := mgr.GetIndexDefnById(common.IndexDefnId(defnRef.DefnId))
		if err != nil {
			logging.Debugf("convertTopologyToIndexInstProtoMsg(): Cannot find definition id = %v. Skip", defnRef.DefnId)
			continue
		}

		// Convert definition to protobuf msg
		defn_proto := convertIndexDefnToProtoMsg(defn)

		// iterate through the index inst for this defnition
		// TODO: Remove CREATED state from the if-stmt
		for _, inst := range defnRef.Instances {
			if common.IndexState(inst.State) == common.INDEX_STATE_READY ||
				common.IndexState(inst.State) == common.INDEX_STATE_INITIAL ||
				common.IndexState(inst.State) == common.INDEX_STATE_CREATED ||
				common.IndexState(inst.State) == common.INDEX_STATE_ACTIVE {
				result = append(result, convertIndexInstToProtoMsg(&inst, defn_proto, port))
			}
		}
	}

	return result, nil
}

// clean up
func cleanup(mgr *manager.IndexManager, t *testing.T) {

	err := mgr.HandleDeleteIndexDDL(common.IndexDefnId(200))
	if err != nil {
		logging.Infof("Error deleting index %s:%s, err=%s", "Default", "coordinator_test", err)
	}
	time.Sleep(time.Duration(1000) * time.Millisecond)
}

func setupInitialData_managerTest(mgr *manager.IndexManager, t *testing.T) {

	// Add a new index definition : 100
	idxDefn := &common.IndexDefn{
		DefnId:          common.IndexDefnId(100),
		Name:            "index_manager_test_100",
		Using:           common.ForestDB,
		Bucket:          "Default",
		IsPrimary:       false,
		SecExprs:        []string{"Testing"},
		ExprType:        common.N1QL,
		PartitionScheme: common.HASH,
		PartitionKey:    "Testing"}

	err := mgr.HandleCreateIndexDDL(idxDefn)
	if err != nil {
		t.Fatal(err)
	}

	// Add a new index definition : 101
	idxDefn = &common.IndexDefn{
		DefnId:          common.IndexDefnId(101),
		Name:            "index_manager_test_101",
		Using:           common.ForestDB,
		Bucket:          "Default",
		IsPrimary:       false,
		SecExprs:        []string{"Testing"},
		ExprType:        common.N1QL,
		PartitionScheme: common.HASH,
		PartitionKey:    "Testing"}

	err = mgr.HandleCreateIndexDDL(idxDefn)
	if err != nil {
		t.Fatal(err)
	}

	err = mgr.UpdateIndexInstance("Default", common.IndexDefnId(101), common.INDEX_STATE_ACTIVE, common.StreamId(0), "")
	if err != nil {
		util.TT.Fatal(err)
	}

	err = mgr.UpdateIndexInstance("Default", common.IndexDefnId(102), common.INDEX_STATE_ACTIVE, common.StreamId(0), "")
	if err != nil {
		util.TT.Fatal(err)
	}
}

func (b *metadataClient) roundRobin(defnID uint64, retry int) uint64 {
	id := common.IndexDefnId(defnID)
	replicas, ok := b.replicas[id]
	if l := len(replicas); ok && l > 0 {
		return uint64(replicas[retry%l])
	}
	return defnID
}

func (m *LifecycleMgr) handleTopologyChange(content []byte) error {

	change := new(topologyChange)
	if err := json.Unmarshal(content, change); err != nil {
		return err
	}

	return m.UpdateIndexInstance(change.Bucket, common.IndexDefnId(change.DefnId), common.IndexState(change.State),
		common.StreamId(change.StreamId), change.Error, change.BuildTime)
}

func deleteIndexForDeleteTest(mgr *manager.IndexManager) {

	//
	// Delete a new index definition : 401 (Default)
	//
	logging.Infof("Run Delete Test : Delete Index Defn 401")
	if err := mgr.HandleDeleteIndexDDL(common.IndexDefnId(401)); err != nil {
		util.TT.Fatal(err)
	}
}

// BuildIndexes implements BridgeAccessor{} interface.
func (b *metadataClient) BuildIndexes(defnIDs []uint64) error {
	_, ok := b.getNodes(defnIDs)
	if !ok {
		return ErrorIndexNotFound
	}
	ids := make([]common.IndexDefnId, len(defnIDs))
	for i, id := range defnIDs {
		ids[i] = common.IndexDefnId(id)
	}
	return b.mdClient.BuildIndexes(ids)
}