// Cursor returns an iterator for a key.
func (tx *Tx) Cursor(key string, direction tsdb.Direction) tsdb.Cursor {
	// Retrieve key bucket.
	b := tx.Bucket([]byte(key))

	tx.engine.mu.RLock()
	defer tx.engine.mu.RUnlock()

	// Ignore if there is no bucket or points in the cache.
	partitionID := WALPartition([]byte(key))
	if b == nil && len(tx.engine.cache[partitionID][key]) == 0 {
		return nil
	}

	// Retrieve a copy of the in-cache points for the key.
	cache := make([][]byte, len(tx.engine.cache[partitionID][key]))
	copy(cache, tx.engine.cache[partitionID][key])

	// Build a cursor that merges the bucket and cache together.
	cur := &Cursor{cache: cache, direction: direction}
	if b != nil {
		cur.cursor = b.Cursor()
	}

	// If it's a reverse cursor, set the current location to the end.
	if direction.Reverse() {
		cur.index = len(cache) - 1
		if cur.cursor != nil {
			cur.cursor.Last()
		}
	}
	return cur
}

// NewCursor returns a new instance of Cursor.
func NewCursor(direction tsdb.Direction, items []CursorItem) *Cursor {
	index := 0
	sort.Sort(CursorItems(items))

	if direction.Reverse() {
		index = len(items)
	}
	return &Cursor{direction: direction, items: items, index: index}
}

func newCursor(cache [][]byte, direction tsdb.Direction) *cursor {
	// position is set such that a call to Next will successfully advance
	// to the next postion and return the value.
	c := &cursor{cache: cache, direction: direction, position: -1}
	if direction.Reverse() {
		c.position = len(c.cache)
	}
	return c
}

// Cursor returns an iterator for a key.
func (tx *Tx) Cursor(key string, direction tsdb.Direction) tsdb.Cursor {
	walCursor := tx.wal.Cursor(key, direction)

	// Retrieve points bucket. Ignore if there is no bucket.
	b := tx.Bucket([]byte("points")).Bucket([]byte(key))
	if b == nil {
		return walCursor
	}

	c := &Cursor{
		cursor:    b.Cursor(),
		direction: direction,
	}

	if direction.Reverse() {
		c.last()
	}

	return tsdb.MultiCursor(direction, walCursor, c)
}