// MakeSplitKey transforms an SQL table key such that it is a valid split key
// (i.e. does not occur in the middle of a row).
func MakeSplitKey(key roachpb.Key) (roachpb.Key, error) {
	if encoding.PeekType(key) != encoding.Int {
		// Not a table key, so already a split key.
		return key, nil
	}

	n := len(key)
	// The column ID length is encoded as a varint and we take advantage of the
	// fact that the column ID itself will be encoded in 0-9 bytes and thus the
	// length of the column ID data will fit in a single byte.
	buf := key[n-1:]
	if encoding.PeekType(buf) != encoding.Int {
		// The last byte is not a valid column ID suffix.
		return nil, util.Errorf("%s: not a valid table key", key)
	}

	// Strip off the column ID suffix from the buf. The last byte of the buf
	// contains the length of the column ID suffix (which might be 0 if the buf
	// does not contain a column ID suffix).
	_, colIDLen, err := encoding.DecodeUvarint(buf)
	if err != nil {
		return nil, err
	}
	if int(colIDLen)+1 > n {
		// The column ID length was impossible. colIDLen is the length of the
		// encoded column ID suffix. We add 1 to account for the byte holding the
		// length of the encoded column ID and if that total (colIDLen+1) is
		// greater than the key suffix (n == len(buf)) then we bail. Note that we
		// don't consider this an error because MakeSplitKey can be called on keys
		// that look like table keys but which do not have a column ID length
		// suffix (e.g SystemConfig.ComputeSplitKeys).
		return nil, util.Errorf("%s: malformed table key", key)
	}
	return key[:len(key)-int(colIDLen)-1], nil
}

func decodeKeyPrint(key roachpb.Key) string {
	var buf bytes.Buffer
	for k := 0; len(key) > 0; k++ {
		var err error
		switch encoding.PeekType(key) {
		case encoding.Null:
			key, _ = encoding.DecodeIfNull(key)
			fmt.Fprintf(&buf, "/NULL")
		case encoding.NotNull:
			key, _ = encoding.DecodeIfNotNull(key)
			fmt.Fprintf(&buf, "/#")
		case encoding.Int:
			var i int64
			key, i, err = encoding.DecodeVarintAscending(key)
			if err == nil {
				fmt.Fprintf(&buf, "/%d", i)
			}
		case encoding.Float:
			var f float64
			key, f, err = encoding.DecodeFloatAscending(key, nil)
			if err == nil {
				fmt.Fprintf(&buf, "/%f", f)
			}
		case encoding.Bytes:
			var s string
			key, s, err = encoding.DecodeStringAscending(key, nil)
			if err == nil {
				fmt.Fprintf(&buf, "/%q", s)
			}
		case encoding.BytesDesc:
			var s string
			key, s, err = encoding.DecodeStringDescending(key, nil)
			if err == nil {
				fmt.Fprintf(&buf, "/%q", s)
			}
		case encoding.Time:
			var t time.Time
			key, t, err = encoding.DecodeTimeAscending(key)
			if err == nil {
				fmt.Fprintf(&buf, "/%s", t.UTC().Format(time.UnixDate))
			}
		case encoding.TimeDesc:
			var t time.Time
			key, t, err = encoding.DecodeTimeDescending(key)
			if err == nil {
				fmt.Fprintf(&buf, "/%s", t.UTC().Format(time.UnixDate))
			}
		default:
			// This shouldn't ever happen, but if it does let the loop exit.
			fmt.Fprintf(&buf, "/%q", []byte(key))
			key = nil
		}

		if err != nil {
			fmt.Fprintf(&buf, "/<%v>", err)
			continue
		}
	}
	return buf.String()
}

// ObjectIDForKey returns the object ID (table or database) for 'key',
// or (_, false) if not within the structured key space.
func ObjectIDForKey(key roachpb.RKey) (uint32, bool) {
	if key.Equal(roachpb.RKeyMax) {
		return 0, false
	}
	if encoding.PeekType(key) != encoding.Int {
		// TODO(marc): this should eventually return SystemDatabaseID.
		return 0, false
	}
	// Consume first encoded int.
	_, id64, err := encoding.DecodeUvarint(key)
	return uint32(id64), err == nil
}

// prettyKey pretty-prints the specified key, skipping over the first skip
// fields.
func prettyKey(key roachpb.Key, skip int) string {
	if !bytes.HasPrefix(key, keys.TableDataPrefix) {
		return fmt.Sprintf("index key missing table data prefix: %q vs %q",
			key, keys.TableDataPrefix)
	}
	key = key[len(keys.TableDataPrefix):]

	var buf bytes.Buffer
	for k := 0; len(key) > 0; k++ {
		var d interface{}
		var err error
		switch encoding.PeekType(key) {
		case encoding.Null:
			key, _ = encoding.DecodeIfNull(key)
			d = parser.DNull
		case encoding.NotNull:
			key, _ = encoding.DecodeIfNotNull(key)
			d = "#"
		case encoding.Int:
			var i int64
			key, i, err = encoding.DecodeVarint(key)
			d = parser.DInt(i)
		case encoding.Float:
			var f float64
			key, f, err = encoding.DecodeFloat(key, nil)
			d = parser.DFloat(f)
		case encoding.Bytes:
			var s string
			key, s, err = encoding.DecodeString(key, nil)
			d = parser.DString(s)
		case encoding.Time:
			var t time.Time
			key, t, err = encoding.DecodeTime(key)
			d = parser.DTimestamp{Time: t}
		default:
			// This shouldn't ever happen, but if it does let the loop exit.
			key = nil
			d = "unknown"
		}
		if skip > 0 {
			skip--
			continue
		}
		if err != nil {
			fmt.Fprintf(&buf, "/<%v>", err)
			continue
		}
		fmt.Fprintf(&buf, "/%s", d)
	}
	return buf.String()
}

func localRangeIDKeyPrint(key roachpb.Key) string {
	var buf bytes.Buffer
	if encoding.PeekType(key) != encoding.Int {
		return fmt.Sprintf("/err<%q>", []byte(key))
	}

	// Get the rangeID.
	key, i, err := encoding.DecodeVarintAscending(key)
	if err != nil {
		return fmt.Sprintf("/err<%v:%q>", err, []byte(key))
	}

	fmt.Fprintf(&buf, "/%d", i)

	// Print and remove the rangeID infix specifier.
	if len(key) != 0 {
		fmt.Fprintf(&buf, "/%s", string(key[0]))
		key = key[1:]
	}

	// Get the suffix.
	hasSuffix := false
	for _, s := range rangeIDSuffixDict {
		if bytes.HasPrefix(key, s.suffix) {
			fmt.Fprintf(&buf, "/%s", s.name)
			key = key[len(s.suffix):]
			if s.ppFunc != nil && len(key) != 0 {
				fmt.Fprintf(&buf, "%s", s.ppFunc(key))
				return buf.String()
			}
			hasSuffix = true
			break
		}
	}

	// Get the encode values.
	if hasSuffix {
		fmt.Fprintf(&buf, "%s", decodeKeyPrint(key))
	} else {
		fmt.Fprintf(&buf, "%q", []byte(key))
	}

	return buf.String()
}

func decodeIndexKeyPrefix(desc *TableDescriptor, key []byte) (IndexID, []byte, error) {
	if encoding.PeekType(key) != encoding.Int {
		return 0, nil, util.Errorf("%s: invalid key prefix: %q", desc.Name, key)
	}

	key, tableID, err := encoding.DecodeUvarintAscending(key)
	if err != nil {
		return 0, nil, err
	}
	key, indexID, err := encoding.DecodeUvarintAscending(key)
	if err != nil {
		return 0, nil, err
	}

	if ID(tableID) != desc.ID {
		return IndexID(indexID), nil, util.Errorf("%s: unexpected table ID: %d != %d", desc.Name, desc.ID, tableID)
	}

	return IndexID(indexID), key, nil
}

func localRangeIDKeyPrint(key roachpb.Key) string {
	var buf bytes.Buffer
	if encoding.PeekType(key) != encoding.Int {
		return fmt.Sprintf("/err<%q>", []byte(key))
	}

	// get range id
	key, i, err := encoding.DecodeVarint(key)
	if err != nil {
		return fmt.Sprintf("/err<%v:%q>", err, []byte(key))
	}

	fmt.Fprintf(&buf, "/%d", i)

	// get suffix

	hasSuffix := false
	for _, s := range rangeIDSuffixDict {
		if bytes.HasPrefix(key, s.suffix) {
			fmt.Fprintf(&buf, "/%s", s.name)
			key = key[len(s.suffix):]
			if s.ppFunc != nil && len(key) != 0 {
				fmt.Fprintf(&buf, "%s", s.ppFunc(key))
				return buf.String()
			}
			hasSuffix = true
			break
		}
	}

	// get encode values
	if hasSuffix {
		fmt.Fprintf(&buf, "%s", decodeKeyPrint(key))
	} else {
		fmt.Fprintf(&buf, "%q", []byte(key))
	}

	return buf.String()
}

// DecodeTablePrefix validates that the given key has a table prefix, returning
// the remainder of the key (with the prefix removed) and the decoded descriptor
// ID of the table.
func DecodeTablePrefix(key roachpb.Key) ([]byte, uint64, error) {
	if encoding.PeekType(key) != encoding.Int {
		return key, 0, util.Errorf("invalid key prefix: %q", key)
	}
	return encoding.DecodeUvarint(key)
}