package pebble

import (
	"bytes"
	"strings"
	"testing"

	"github.com/cockroachdb/pebble/internal/base"
	"github.com/cockroachdb/pebble/internal/keyspan"
	"github.com/cockroachdb/pebble/internal/testkeys"
	"github.com/cockroachdb/pebble/sstable"
	"github.com/cockroachdb/pebble/vfs"
	"github.com/stretchr/testify/require"
)

func Test(t *testing.T) {
	equal := func(_ base.Compare, a, b keyspan.Span) bool { return true }
	transform := func(cmp base.Compare, in keyspan.Span, out *keyspan.Span) error {
		out.Start, out.End = in.Start, in.End
		out.Keys = append(out.Keys[:0], in.Keys...)
		return nil
	}

	var keys = []string{
		"a b",
		"b c",
		"c d",
		"x y",
		"y z",
	}

	// Create a table.
	fs := vfs.NewMem()
	f, err := fs.Create("test.sst")
	require.NoError(t, err)
	w := sstable.NewWriter(f, sstable.WriterOptions{TableFormat: sstable.TableFormatMax})

	for _, k := range keys {
		parts := strings.Split(k, " ")
		start, end := []byte(parts[0]), []byte(parts[1])
		err = w.Add(base.MakeInternalKey(start, 0, base.InternalKeyKindRangeDelete), end)
		require.NoError(t, err)
	}
	err = w.Close()
	require.NoError(t, err)

	// Construct an iterator over range keys in the table.
	f, err = fs.Open("test.sst")
	require.NoError(t, err)
	r, err := sstable.NewReader(f, sstable.ReaderOptions{})
	require.NoError(t, err)

	iter, err := r.NewRawRangeDelIter()
	require.NoError(t, err)

	cmp := testkeys.Comparer.Compare
	dIter := &keyspan.DefragmentingIter{}
	dIter.Init(cmp, iter, equal, keyspan.StaticDefragmentReducer)
	iter = dIter

	_ = transform
	// Using a merging iter results in a panic.
	//mIter := &keyspan.MergingIter{}
	//mIter.Init(cmp, transform, iter)
	//iter = mIter

	var b bytes.Buffer
	for s := iter.First(); s.Valid(); s = iter.Next() {
		b.WriteString(s.String() + "\n")
	}
	err = iter.Close()
	require.NoError(t, err)

	const want = `
a-d:{(#0,RANGEDEL)}
x-z:{(#0,RANGEDEL)}
`
	require.Equal(t, strings.TrimSpace(want), strings.TrimSpace(b.String()))
}