// ExampleSetAndGetZone sets zone configs for a variety of key
// prefixes and verifies they can be fetched directly.
func ExampleSetAndGetZone() {
	httpServer := startAdminServer()
	defer httpServer.Close()
	testConfigFn := createTestConfigFile()
	defer os.Remove(testConfigFn)

	testData := []struct {
		prefix storage.Key
		yaml   string
	}{
		{storage.KeyMin, testConfig},
		{storage.Key("db1"), testConfig},
		{storage.Key("db 2"), testConfig},
		{storage.Key("\xfe"), testConfig},
	}

	for _, test := range testData {
		prefix := url.QueryEscape(string(test.prefix))
		runSetZone(CmdSetZone, []string{prefix, testConfigFn})
		runGetZones(CmdGetZone, []string{prefix})
	}
	// Output:
	// set zone config for key prefix ""
	// zone config for key prefix "":
	//
	// replicas:
	//   dc1:
	//   - SSD
	// range_min_bytes: 1048576
	// range_max_bytes: 67108864
	//
	// set zone config for key prefix "db1"
	// zone config for key prefix "db1":
	//
	// replicas:
	//   dc1:
	//   - SSD
	// range_min_bytes: 1048576
	// range_max_bytes: 67108864
	//
	// set zone config for key prefix "db+2"
	// zone config for key prefix "db+2":
	//
	// replicas:
	//   dc1:
	//   - SSD
	// range_min_bytes: 1048576
	// range_max_bytes: 67108864
	//
	// set zone config for key prefix "%FE"
	// zone config for key prefix "%FE":
	//
	// replicas:
	//   dc1:
	//   - SSD
	// range_min_bytes: 1048576
	// range_max_bytes: 67108864
}

// ExampleLsZones creates a series of zone configs and verifies
// zone-ls works. First, no regexp lists all zone configs. Second,
// regexp properly matches results.
func ExampleLsZones() {
	httpServer := startAdminServer()
	defer httpServer.Close()
	testConfigFn := createTestConfigFile()
	defer os.Remove(testConfigFn)

	keys := []storage.Key{
		storage.KeyMin,
		storage.Key("db1"),
		storage.Key("db2"),
		storage.Key("db3"),
		storage.Key("user"),
	}

	regexps := []string{
		"",
		"db*",
		"db[12]",
	}

	for _, key := range keys {
		prefix := url.QueryEscape(string(key))
		runSetZone(CmdSetZone, []string{prefix, testConfigFn})
	}

	for i, regexp := range regexps {
		fmt.Fprintf(os.Stdout, "test case %d: %q\n", i, regexp)
		if regexp == "" {
			runLsZones(CmdLsZones, []string{})
		} else {
			runLsZones(CmdLsZones, []string{regexp})
		}
	}
	// Output:
	// set zone config for key prefix ""
	// set zone config for key prefix "db1"
	// set zone config for key prefix "db2"
	// set zone config for key prefix "db3"
	// set zone config for key prefix "user"
	// test case 0: ""
	// [default]
	// db1
	// db2
	// db3
	// user
	// test case 1: "db*"
	// db1
	// db2
	// db3
	// test case 2: "db[12]"
	// db1
	// db2
}

func TestKeysAndBodyArePreserved(t *testing.T) {
	encKey := "%00some%2Fkey%20that%20encodes%E4%B8%96%E7%95%8C"
	encBody := "%00some%2FBODY%20that%20encodes"
	s := runHTTPTestFixture(t, []RequestResponse{
		{
			NewRequest("POST", encKey, encBody),
			NewResponse(200),
		},
		{
			NewRequest("GET", encKey),
			NewResponse(200, encBody, "application/octet-stream"),
		},
	})
	gr := <-s.db.Get(&storage.GetRequest{
		RequestHeader: storage.RequestHeader{
			Key:  storage.Key("\x00some/key that encodes世界"),
			User: storage.UserRoot,
		},
	})
	if gr.Error != nil {
		t.Errorf("Unable to fetch values from local db")
	}
	if !bytes.Equal(gr.Value.Bytes, []byte(encBody)) {
		t.Errorf("Expected value (%s) but got (%s)", encBody, gr.Value.Bytes)
	}
}

func dbKey(path string) (storage.Key, error) {
	result, err := url.QueryUnescape(strings.TrimPrefix(path, KVKeyPrefix))
	if err == nil {
		return storage.Key(result), nil
	}
	return nil, err
}

// ExampleRmZones creates a series of zone configs and verifies
// zone-rm works by deleting some and then all and verifying entries
// have been removed via zone-ls. Also verify the default zone cannot
// be removed.
func ExampleRmZones() {
	httpServer := startAdminServer()
	defer httpServer.Close()
	testConfigFn := createTestConfigFile()
	defer os.Remove(testConfigFn)

	keys := []storage.Key{
		storage.KeyMin,
		storage.Key("db1"),
	}

	for _, key := range keys {
		prefix := url.QueryEscape(string(key))
		runSetZone(CmdSetZone, []string{prefix, testConfigFn})
	}

	for _, key := range keys {
		prefix := url.QueryEscape(string(key))
		runRmZone(CmdRmZone, []string{prefix})
		runLsZones(CmdLsZones, []string{})
	}
	// Output:
	// set zone config for key prefix ""
	// set zone config for key prefix "db1"
	// [default]
	// db1
	// removed zone config for key prefix "db1"
	// [default]
}

// TestBootstrap verifies the results of bootstrapping a cluster. Uses
// an in memory engine.
func TestBootstrapCluster(t *testing.T) {
	engine := storage.NewInMem(1 << 20)
	localDB, err := BootstrapCluster("cluster-1", engine)
	if err != nil {
		t.Fatal(err)
	}
	// Scan the complete contents of the local database.
	sr := <-localDB.Scan(&storage.ScanRequest{
		StartKey:   storage.KeyMin,
		EndKey:     storage.KeyMax,
		MaxResults: math.MaxInt64,
	})
	if sr.Error != nil {
		t.Fatal(sr.Error)
	}
	var keys []storage.Key
	for _, kv := range sr.Rows {
		keys = append(keys, kv.Key)
	}
	var expectedKeys = []storage.Key{
		storage.Key("\x00\x00\x00range-1"),
		storage.Key("\x00\x00\x00range-id-generator"),
		storage.Key("\x00\x00\x00store-ident"),
		storage.Key("\x00\x00meta1\xff"),
		storage.Key("\x00\x00meta2\xff"),
		storage.Key("\x00node-id-generator"),
		storage.Key("\x00store-id-generator-1"),
	}
	if !reflect.DeepEqual(keys, expectedKeys) {
		t.Errorf("expected keys mismatch:\n%s\n  -- vs. -- \n\n%s",
			formatKeys(keys), formatKeys(expectedKeys))
	}

	// TODO(spencer): check values.
}

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

func dbKey(path, apiPrefix string) (storage.Key, error) {
	result, err := url.QueryUnescape(strings.TrimPrefix(path, apiPrefix))
	if err == nil {
		k := storage.Key(result)
		if len(k) == 0 {
			return nil, fmt.Errorf("empty key not allowed")
		}
		return k, nil
	}
	return nil, err
}

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

func TestReplicaLookup(t *testing.T) {

	db := NewLocalDB()
	r1 := db.addTestRange(storage.KeyMin, storage.Key("C"))
	r2 := db.addTestRange(storage.Key("C"), storage.Key("X"))
	r3 := db.addTestRange(storage.Key("X"), storage.KeyMax)
	if len(db.ranges) != 3 {
		t.Errorf("Pre-condition failed! Expected ranges to be size 3, got %d", len(db.ranges))
	}

	assertReplicaForRange(t, db.lookupReplica(storage.KeyMin), r1)
	assertReplicaForRange(t, db.lookupReplica(storage.Key("B")), r1)
	assertReplicaForRange(t, db.lookupReplica(storage.Key("C")), r2)
	assertReplicaForRange(t, db.lookupReplica(storage.Key("M")), r2)
	assertReplicaForRange(t, db.lookupReplica(storage.Key("X")), r3)
	assertReplicaForRange(t, db.lookupReplica(storage.Key("Z")), r3)
	if db.lookupReplica(storage.KeyMax) != nil {
		t.Errorf("Expected storage.KeyMax to not have an associated Replica.")
	}
}

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