// BootstrapRangeDescriptor sets meta1 and meta2 values for KeyMax,
// using the provided replica.
func BootstrapRangeDescriptor(db DB, replica storage.Replica) error {
	locations := storage.RangeDescriptor{
		StartKey: storage.KeyMin,
		Replicas: []storage.Replica{replica},
	}
	// Write meta1.
	if err := PutI(db, storage.MakeKey(storage.KeyMeta1Prefix, storage.KeyMax), locations); err != nil {
		return err
	}
	// Write meta2.
	if err := PutI(db, storage.MakeKey(storage.KeyMeta2Prefix, storage.KeyMax), locations); err != nil {
		return err
	}
	return nil
}

// BootstrapRangeLocations sets meta1 and meta2 values for KeyMax,
// using the provided replica.
func BootstrapRangeLocations(db DB, replica storage.Replica) error {
	locations := storage.RangeLocations{
		Replicas: []storage.Replica{replica},
		// TODO(spencer): uncomment when we have hrsht's change.
		//StartKey: storage.KeyMin,
	}
	// Write meta1.
	if err := PutI(db, storage.MakeKey(storage.KeyMeta1Prefix, storage.KeyMax), locations); err != nil {
		return err
	}
	// Write meta2.
	if err := PutI(db, storage.MakeKey(storage.KeyMeta2Prefix, storage.KeyMax), locations); err != nil {
		return err
	}
	return nil
}

// TODO(harshit): Consider caching returned metadata info.
func (db *DistDB) lookupMeta1(key storage.Key) (*storage.RangeLocations, error) {
	info, err := db.gossip.GetInfo(gossip.KeyFirstRangeMetadata)
	if err != nil {
		return nil, err
	}
	metadataKey := storage.MakeKey(storage.KeyMeta1Prefix, key)
	return db.lookupMetadata(metadataKey, info.(storage.RangeLocations).Replicas)
}

func (db *DistDB) lookupMeta2(key storage.Key) (*storage.RangeLocations, error) {
	meta1Val, err := db.lookupMeta1(key)
	if err != nil {
		return nil, err
	}
	metadataKey := storage.MakeKey(storage.KeyMeta2Prefix, key)
	return db.lookupMetadata(metadataKey, meta1Val.Replicas)
}

// Get retrieves the zone configuration for the specified key. If the
// key is empty, all zone configurations are returned. Otherwise, the
// leading "/" path delimiter is stripped and the zone configuration
// matching the remainder is retrieved. Note that this will retrieve
// the default zone config if "key" is equal to "/", and will list all
// configs if "key" is equal to "". The body result contains
// JSON-formatted output for a listing of keys and YAML-formatted
// output for retrieval of a zone config.
func (zh *zoneHandler) Get(path string, r *http.Request) (body []byte, contentType string, err error) {
	// Scan all zones if the key is empty.
	if len(path) == 0 {
		sr := <-zh.kvDB.Scan(&storage.ScanRequest{
			RequestHeader: storage.RequestHeader{
				Key:    storage.KeyConfigZonePrefix,
				EndKey: storage.PrefixEndKey(storage.KeyConfigZonePrefix),
				User:   storage.UserRoot,
			},
			MaxResults: maxGetResults,
		})
		if sr.Error != nil {
			err = sr.Error
			return
		}
		if len(sr.Rows) == maxGetResults {
			glog.Warningf("retrieved maximum number of results (%d); some may be missing", maxGetResults)
		}
		var prefixes []string
		for _, kv := range sr.Rows {
			trimmed := bytes.TrimPrefix(kv.Key, storage.KeyConfigZonePrefix)
			prefixes = append(prefixes, url.QueryEscape(string(trimmed)))
		}
		// JSON-encode the prefixes array.
		contentType = "application/json"
		if body, err = json.Marshal(prefixes); err != nil {
			err = util.Errorf("unable to format zone configurations: %v", err)
		}
	} else {
		zoneKey := storage.MakeKey(storage.KeyConfigZonePrefix, storage.Key(path[1:]))
		var ok bool
		config := &storage.ZoneConfig{}
		if ok, _, err = kv.GetI(zh.kvDB, zoneKey, config); err != nil {
			return
		}
		// On get, if there's no zone config for the requested prefix,
		// return a not found error.
		if !ok {
			err = util.Errorf("no config found for key prefix %q", path)
			return
		}
		var out []byte
		if out, err = yaml.Marshal(config); err != nil {
			err = util.Errorf("unable to marshal zone config %+v to yaml: %v", config, err)
			return
		}
		if !utf8.ValidString(string(out)) {
			err = util.Errorf("config contents not valid utf8: %q", out)
			return
		}
		contentType = "text/yaml"
		body = out
	}

	return
}

// UpdateRangeDescriptor updates the range locations metadata for the
// range specified by the meta parameter. This always involves a write
// to "meta2", and may require a write to "meta1", in the event that
// meta.EndKey is a "meta2" key (prefixed by KeyMeta2Prefix).
func UpdateRangeDescriptor(db DB, meta storage.RangeMetadata,
	desc storage.RangeDescriptor, timestamp hlc.HLTimestamp) error {
	// TODO(spencer): a lot more work here to actually implement this.

	// Write meta2.
	key := storage.MakeKey(storage.KeyMeta2Prefix, meta.EndKey)
	if err := PutI(db, key, desc, timestamp); err != nil {
		return err
	}
	return nil
}

// allocateStoreIDs increments the store id generator key for the
// specified node to allocate "inc" new, unique store ids. The
// first ID in a contiguous range is returned on success.
func allocateStoreIDs(nodeID int32, inc int64, db kv.DB) (int32, error) {
	ir := <-db.Increment(&storage.IncrementRequest{
		// The Key is a concatenation of StoreIDGeneratorPrefix and this node's ID.
		Key: storage.MakeKey(storage.KeyStoreIDGeneratorPrefix,
			[]byte(strconv.Itoa(int(nodeID)))),
		Increment: inc,
	})
	if ir.Error != nil {
		return 0, util.Errorf("unable to allocate %d store IDs for node %d: %v", inc, nodeID, ir.Error)
	}
	return int32(ir.NewValue - inc + 1), nil
}

// Delete removes the zone config specified by key.
func (zh *zoneHandler) Delete(path string, r *http.Request) error {
	if len(path) == 0 {
		return util.Errorf("no path specified for zone Delete")
	}
	if path == "/" {
		return util.Errorf("the default zone configuration cannot be deleted")
	}
	zoneKey := storage.MakeKey(storage.KeyConfigZonePrefix, storage.Key(path[1:]))
	dr := <-zh.kvDB.Delete(&storage.DeleteRequest{Key: zoneKey})
	if dr.Error != nil {
		return dr.Error
	}
	return nil
}

// lookupRangeMetadata first looks up the specified key in the first
// level of range metadata and then looks up the specified key in the
// second level of range metadata to yield the set of replicas where
// the key resides. This process is retried in a loop until the key's
// replicas are located or a non-retryable error is encountered.
func (db *DistDB) lookupRangeMetadata(key storage.Key) (*storage.RangeDescriptor, error) {
	firstLevelMeta, err := db.lookupRangeMetadataFirstLevel(key)
	if err != nil {
		return nil, err
	}
	metadataKey := storage.MakeKey(storage.KeyMeta2Prefix, key)
	args := &storage.InternalRangeLookupRequest{Key: metadataKey}
	replyChan := make(chan *storage.InternalRangeLookupResponse, len(firstLevelMeta.Replicas))
	if err = db.sendRPC(firstLevelMeta.Replicas, "Node.InternalRangeLookup", args, replyChan); err != nil {
		return nil, err
	}
	reply := <-replyChan
	return &reply.Range, nil
}

// lookupRangeMetadataFirstLevel issues an InternalRangeLookup request
// to the first-level range metadata table. This always chooses from
// amongst the first range metadata replicas (these are gossipped).
func (db *DistDB) lookupRangeMetadataFirstLevel(key storage.Key) (*storage.RangeDescriptor, error) {
	info, err := db.gossip.GetInfo(gossip.KeyFirstRangeMetadata)
	if err != nil {
		return nil, firstRangeMissingErr{err}
	}
	replicas := info.(storage.RangeDescriptor).Replicas
	metadataKey := storage.MakeKey(storage.KeyMeta1Prefix, key)
	args := &storage.InternalRangeLookupRequest{Key: metadataKey}
	replyChan := make(chan *storage.InternalRangeLookupResponse, len(replicas))
	if err = db.sendRPC(replicas, "Node.InternalRangeLookup", args, replyChan); err != nil {
		return nil, err
	}
	reply := <-replyChan
	return &reply.Range, nil
}

// Get retrieves the zone configuration for the specified key. If the
// key is empty, all zone configurations are returned. Otherwise, the
// leading "/" path delimiter is stripped and the zone configuration
// matching the remainder is retrieved. Note that this will retrieve
// the default zone config if "key" is equal to "/", and will list all
// configs if "key" is equal to "". The body result contains
// JSON-formmatted output via the GetZoneResponse struct.
func (zh *zoneHandler) Get(path string, r *http.Request) (body []byte, contentType string, err error) {
	contentType = "application/json"

	// Scan all zones if the key is empty.
	if len(path) == 0 {
		sr := <-zh.kvDB.Scan(&storage.ScanRequest{
			StartKey:   storage.KeyConfigZonePrefix,
			EndKey:     storage.PrefixEndKey(storage.KeyConfigZonePrefix),
			MaxResults: maxGetResults,
		})
		if sr.Error != nil {
			err = sr.Error
			return
		}
		if len(sr.Rows) == maxGetResults {
			glog.Warningf("retrieved maximum number of results (%d); some may be missing", maxGetResults)
		}
		var prefixes []string
		for _, kv := range sr.Rows {
			trimmed := bytes.TrimPrefix(kv.Key, storage.KeyConfigZonePrefix)
			prefixes = append(prefixes, url.QueryEscape(string(trimmed)))
		}
		// JSON-encode the prefixes array.
		if body, err = json.Marshal(prefixes); err != nil {
			err = util.Errorf("unable to format zone configurations: %v", err)
		}
	} else {
		zoneKey := storage.MakeKey(storage.KeyConfigZonePrefix, storage.Key(path[1:]))
		gr := <-zh.kvDB.Get(&storage.GetRequest{Key: zoneKey})
		if gr.Error != nil {
			return
		}
		// On get, if there's no zone config for the requested prefix,
		// return a not found error.
		if gr.Value.Bytes == nil {
			err = util.Errorf("no config found for key prefix %q", path)
			return
		}
		if !utf8.ValidString(string(gr.Value.Bytes)) {
			err = util.Errorf("config contents not valid utf8: %q", gr.Value)
			return
		}
		body = gr.Value.Bytes
	}

	return
}

// Put writes a zone config for the specified key prefix "key".  The
// zone config is parsed from the input "body". The zone config is
// stored gob-encoded. The specified body must be valid utf8 and must
// validly parse into a zone config struct.
func (zh *zoneHandler) Put(path string, body []byte, r *http.Request) error {
	if len(path) == 0 {
		return util.Errorf("no path specified for zone Put")
	}
	configStr := string(body)
	if !utf8.ValidString(configStr) {
		return util.Errorf("config contents not valid utf8: %q", body)
	}
	config, err := storage.ParseZoneConfig(body)
	if err != nil {
		return util.Errorf("zone config has invalid format: %s: %v", configStr, err)
	}
	zoneKey := storage.MakeKey(storage.KeyConfigZonePrefix, storage.Key(path[1:]))
	if err := kv.PutI(zh.kvDB, zoneKey, config, hlc.HLTimestamp{}); err != nil {
		return err
	}
	return nil
}

// Put writes a zone config for the specified key prefix "key".  The
// zone config is parsed from the input "body". The zone config is
// stored as YAML text. The specified body must be valid utf8 and
// must validly parse into a zone config struct.
func (zh *zoneHandler) Put(path string, body []byte, r *http.Request) error {
	if len(path) == 0 {
		return util.Errorf("no path specified for zone Put")
	}
	configStr := string(body)
	if !utf8.ValidString(configStr) {
		return util.Errorf("config contents not valid utf8: %q", body)
	}
	_, err := storage.ParseZoneConfig(body)
	if err != nil {
		return util.Errorf("zone config has invalid format: %s: %v", configStr, err)
	}
	zoneKey := storage.MakeKey(storage.KeyConfigZonePrefix, storage.Key(path[1:]))
	pr := <-zh.kvDB.Put(&storage.PutRequest{Key: zoneKey, Value: storage.Value{Bytes: body}})
	if pr.Error != nil {
		return pr.Error
	}
	return nil
}

// TestNullPrefixedKeys makes sure that the internal system keys are not accessible through the HTTP API.
func TestNullPrefixedKeys(t *testing.T) {
	// TODO(zbrock + matthew) fix this once sqlite key encoding is finished so we can namespace user keys
	t.Skip("Internal Meta1 Keys should not be accessible from the HTTP REST API. But they are right now.")

	metaKey := storage.MakeKey(storage.KeyMeta1Prefix, storage.KeyMax)
	s := startNewServer()

	// Precondition: we want to make sure the meta1 key exists.
	initialVal, err := s.rawGet(metaKey)
	if err != nil {
		t.Fatalf("Precondition Failed! Unable to fetch %+v from local db", metaKey)
	}
	if initialVal == nil {
		t.Fatalf("Precondition Failed! Expected meta1 key to exist in the underlying store, but no value found")
	}

	// Try to manipulate the meta1 key.
	encMeta1Key := url.QueryEscape(string(metaKey))
	runHTTPTestFixture(t, []RequestResponse{
		{
			NewRequest("GET", encMeta1Key),
			NewResponse(404),
		},
		{
			NewRequest("POST", encMeta1Key, "cool"),
			NewResponse(200),
		},
		{
			NewRequest("GET", encMeta1Key),
			NewResponse(200, "cool", "application/octet-stream"),
		},
	}, s)

	// Postcondition: the meta1 key is untouched.
	afterVal, err := s.rawGet(metaKey)
	if err != nil {
		t.Errorf("Unable to fetch %+v from local db", metaKey)
	}
	if !bytes.Equal(afterVal, initialVal) {
		t.Errorf("Expected meta1 to be unchanged, but it differed: %+v", afterVal)
	}
}