package sorted

import (
	"bytes"
	"os"
	"sort"
	"testing"
)

// Records are 8 bytes: a 4-byte key followed by a 4-byte payload. cmpLen is 4,
// so the payload is invisible to ordering but must survive a round trip.
const (
	tRecLen = 8
	tCmpLen = 4
)

type sortedSink struct {
	recs [][]byte
	n    int32
	max  int32 // 0 = unlimited; otherwise stop after max records
}

func newSortedSink() (s *sortedSink) {
	return &sortedSink{recs: [][]byte{:64}}
}

func (s *sortedSink) add(rec []byte) (ok bool) {
	if s.max > 0 && s.n >= s.max {
		return false
	}
	if s.n >= int32(len(s.recs)) {
		return false
	}
	cp := []byte{:len(rec)}
	copy(cp, rec)
	s.recs[s.n] = cp
	s.n++
	return true
}

func srRec(key byte, payload byte) (r []byte) {
	r = []byte{:tRecLen}
	r[0], r[1], r[2], r[3] = key, key, key, key
	r[4], r[5], r[6], r[7] = payload, payload, payload, payload
	return
}

// srBound is a scan boundary: exactly cmpLen bytes, as Scan's comparisons are
// on the full key slice it is handed, not on key[:cmpLen].
func srBound(key byte) (r []byte) {
	r = []byte{:tCmpLen}
	r[0], r[1], r[2], r[3] = key, key, key, key
	return
}

func srKey(k0, k1, k2, k3 byte, payload byte) (r []byte) {
	r = []byte{:tRecLen}
	r[0], r[1], r[2], r[3] = k0, k1, k2, k3
	r[4], r[5], r[6], r[7] = payload, payload, payload, payload
	return
}

func srTmp(t *testing.T) (dir string, ok bool) {
	t.Helper()
	d, err := os.MkdirTemp("", "sorted-*")
	if err != nil {
		t.Fatal(err)
		return "", false
	}
	return d, true
}

func srOpen(t *testing.T, path string) (f *File, ok bool) {
	t.Helper()
	g, err := Open(path, tRecLen, tCmpLen)
	if err != nil {
		t.Fatal(err)
		return nil, false
	}
	return g, true
}

func TestEmptyFile(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	if f.Count() != 0 {
		t.Fatalf("Count = %d", f.Count())
	}
	if f.Dirty() {
		t.Fatal("a fresh file is not dirty")
	}
	key := srRec(1, 0)
	if _, got := f.Get(key); got {
		t.Fatal("Get on an empty file")
	}
	if _, got := f.GetPrefix(key[:2]); got {
		t.Fatal("GetPrefix on an empty file")
	}
	if _, got := f.Last(); got {
		t.Fatal("Last on an empty file")
	}
	if f.LastFlushedSer() != 0 {
		t.Fatal("LastFlushedSer on a fresh file")
	}
	sink := newSortedSink()
	f.Scan(srBound(0), srBound(255), sink.add)
	if sink.n != 0 {
		t.Fatalf("Scan on an empty file yielded %d", sink.n)
	}
	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	if err := f.Clear(); err != nil {
		t.Fatal(err)
		return
	}
}

func TestPutGetScan(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	f.Put(srRec(3, 0x33))
	f.Put(srRec(1, 0x11))
	f.Put(srRec(2, 0x22))
	if f.Count() != 3 {
		t.Fatalf("Count = %d", f.Count())
	}
	if !f.Dirty() {
		t.Fatal("unflushed records must make the file dirty")
	}

	rec, found := f.Get(srRec(2, 0x99))
	if !found {
		t.Fatal("Get key 2")
	}
	if rec[4] != 0x22 {
		t.Fatalf("Get returned payload %d", rec[4])
	}
	if _, got := f.Get(srRec(9, 0)); got {
		t.Fatal("Get of an absent key")
	}

	last, lok := f.Last()
	if !lok || last[0] != 3 {
		t.Fatal("Last must be the largest key")
	}

	sink := newSortedSink()
	f.Scan(srBound(0), srBound(255), sink.add)
	if sink.n != 3 {
		t.Fatalf("Scan count = %d", sink.n)
	}
	if sink.recs[0][0] != 1 || sink.recs[1][0] != 2 || sink.recs[2][0] != 3 {
		t.Fatal("Scan must return ascending keys")
	}
	if sink.recs[1][4] != 0x22 {
		t.Fatal("Scan lost a payload")
	}

	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	if f.Dirty() {
		t.Fatal("file still dirty after Flush")
	}
	if f.Count() != 3 {
		t.Fatalf("Count after Flush = %d", f.Count())
	}
	diskRec, dfound := f.Get(srRec(2, 0))
	if !dfound || diskRec[4] != 0x22 {
		t.Fatal("Get from disk after Flush")
	}
}

func TestReopenPersists(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f1, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	f1.Put(srRec(1, 0x0A))
	f1.Put(srRec(2, 0x0B))
	if err := f1.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	if err := f1.Close(); err != nil {
		t.Fatal(err)
		return
	}

	f2, ok3 := srOpen(t, dir|"/idx.dat")
	if !ok3 {
		return
	}
	defer f2.Close()
	if f2.Count() != 2 {
		t.Fatalf("reopened Count = %d", f2.Count())
	}
	rec, found := f2.Get(srRec(1, 0))
	if !found || rec[4] != 0x0A {
		t.Fatal("reopened Get key 1")
	}
	last, lok := f2.Last()
	if !lok || last[0] != 2 {
		t.Fatal("reopened Last")
	}
	sink := newSortedSink()
	f2.Scan(srBound(0), srBound(255), sink.add)
	if sink.n != 2 {
		t.Fatalf("reopened Scan count = %d", sink.n)
	}
}

// TestSidecarAcrossHandles exercises the .buf sidecar path: one handle flushes
// records that a second handle opened earlier can only see by re-reading the
// sidecar size.
func TestSidecarAcrossHandles(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f1, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	f2, ok3 := srOpen(t, dir|"/idx.dat")
	if !ok3 {
		f1.Close()
		return
	}
	defer f1.Close()
	defer f2.Close()

	f1.Put(srRec(5, 0x55))
	needs, qerr := f1.QuickFlush(1, false)
	if qerr != nil {
		t.Fatal(qerr)
		return
	}
	if needs {
		t.Fatal("a small sidecar must not ask for a merge")
	}
	if f1.LastFlushedSer() != 1 {
		t.Fatalf("LastFlushedSer = %d", f1.LastFlushedSer())
	}
	if f1.Count() != 1 {
		t.Fatalf("writer Count = %d", f1.Count())
	}
	// The reader sees the sidecar record; Get refreshes the cached sidecar
	// size, so Count is accurate after it.
	rec, found := f2.Get(srRec(5, 0))
	if !found || rec[4] != 0x55 {
		t.Fatal("reader Get from sidecar")
	}
	if f2.Count() != 1 {
		t.Fatalf("reader Count = %d", f2.Count())
	}
	sink := newSortedSink()
	f2.Scan(srBound(0), srBound(255), sink.add)
	if sink.n != 1 || sink.recs[0][0] != 5 {
		t.Fatalf("reader Scan count = %d", sink.n)
	}

	// A second flush grows the sidecar; the reader's cached size refreshes.
	f1.Put(srRec(6, 0x66))
	if _, qerr2 := f1.QuickFlush(2, false); qerr2 != nil {
		t.Fatal(qerr2)
		return
	}
	rec2, found2 := f2.Get(srRec(6, 0))
	if !found2 || rec2[4] != 0x66 {
		t.Fatal("reader Get after a second sidecar flush")
	}
	sink2 := newSortedSink()
	f2.Scan(srBound(0), srBound(255), sink2.add)
	if sink2.n != 2 || sink2.recs[0][0] != 5 || sink2.recs[1][0] != 6 {
		t.Fatalf("reader Scan after a second flush = %d", sink2.n)
	}
}

// TestMixedSourcesDedup has one record on disk, one in the sidecar and two in
// memory, with a duplicate key in two sources. Scan must merge all three
// sources in order and emit a duplicated key once.
func TestMixedSourcesDedup(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	f.Put(srRec(1, 0x01))
	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	f.Put(srRec(2, 0x02))
	if _, qerr := f.QuickFlush(2, false); qerr != nil {
		t.Fatal(qerr)
		return
	}
	f.Put(srRec(3, 0x03))
	f.Put(srRec(1, 0x99)) // duplicate key now in the memory buffer

	sink := newSortedSink()
	f.Scan(srBound(0), srBound(255), sink.add)
	if sink.n != 3 {
		t.Fatalf("merged Scan count = %d", sink.n)
	}
	if sink.recs[0][0] != 1 || sink.recs[1][0] != 2 || sink.recs[2][0] != 3 {
		t.Fatal("merged Scan not ascending")
	}

	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	if f.Count() != 3 {
		t.Fatalf("Count after merge = %d", f.Count())
	}
}

func TestDelete(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	f.Put(srRec(1, 0x01))
	f.Put(srRec(2, 0x02))
	f.Put(srRec(3, 0x03))
	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	delKey := srRec(2, 0)
	f.Delete(delKey)
	if !f.Dirty() {
		t.Fatal("a pending delete must make the file dirty")
	}
	sink := newSortedSink()
	f.Scan(srBound(0), srBound(255), sink.add)
	if sink.n != 2 || sink.recs[0][0] != 1 || sink.recs[1][0] != 3 {
		t.Fatalf("Scan after Delete count = %d", sink.n)
	}

	// Before the Flush compacts the file, the tombstone has to hide the record
	// from a point lookup as well: Get read the on-disk record and never
	// consulted the delete set, so a deleted key was still readable here while
	// Scan already hid it.
	if _, stillThere := f.Get(delKey); stillThere {
		t.Fatal("deleted key still readable before Flush")
	}
	if _, keep := f.Get(srRec(1, 0)); !keep {
		t.Fatal("Get lost a live key while skipping a tombstone")
	}

	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	if _, found := f.Get(delKey); found {
		t.Fatal("deleted key still readable after Flush")
	}
	sink2 := newSortedSink()
	f.Scan(srBound(0), srBound(255), sink2.add)
	if sink2.n != 2 {
		t.Fatalf("Scan after delete Flush count = %d", sink2.n)
	}
	last, lok := f.Last()
	if !lok || last[0] != 3 {
		t.Fatal("Last after delete")
	}
}

func TestQuickFlushEmptyAndThreshold(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	needs, err := f.QuickFlush(7, false)
	if err != nil {
		t.Fatal(err)
		return
	}
	if needs {
		t.Fatal("an empty flush must not ask for a merge")
	}
	if f.LastFlushedSer() != 7 {
		t.Fatalf("LastFlushedSer = %d", f.LastFlushedSer())
	}
	f.Put(srRec(1, 0x01))
	needs2, err2 := f.QuickFlush(9, true)
	if err2 != nil {
		t.Fatal(err2)
		return
	}
	if needs2 {
		t.Fatal("below the 1MB threshold no merge is needed")
	}
	if f.LastFlushedSer() != 9 {
		t.Fatalf("LastFlushedSer after flush = %d", f.LastFlushedSer())
	}
	if f.Count() != 1 {
		t.Fatalf("Count after sidecar flush = %d", f.Count())
	}
}

func TestScanBoundsInclusive(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	for k := 1; k <= 5; k++ {
		f.Put(srRec(byte(k), byte(k)))
	}
	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}

	sink := newSortedSink()
	f.Scan(srBound(2), srBound(4), sink.add)
	if sink.n != 3 || sink.recs[0][0] != 2 || sink.recs[2][0] != 4 {
		t.Fatalf("bounded Scan count = %d", sink.n)
	}

	sink2 := newSortedSink()
	f.Scan(srBound(5), srBound(5), sink2.add)
	if sink2.n != 1 || sink2.recs[0][0] != 5 {
		t.Fatalf("single-key Scan count = %d", sink2.n)
	}

	sink3 := newSortedSink()
	f.Scan(srBound(6), srBound(9), sink3.add)
	if sink3.n != 0 {
		t.Fatalf("out-of-range Scan count = %d", sink3.n)
	}

	sink4 := newSortedSink()
	f.Scan(srBound(0), srBound(0), sink4.add)
	if sink4.n != 0 {
		t.Fatalf("below-range Scan count = %d", sink4.n)
	}
}

func TestScanStopsOnFalse(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	for k := 1; k <= 4; k++ {
		f.Put(srRec(byte(k), byte(k)))
	}
	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	stop := newSortedSink()
	stop.max = 2
	f.Scan(srBound(0), srBound(255), stop.add)
	if stop.n != 2 || stop.recs[0][0] != 1 || stop.recs[1][0] != 2 {
		t.Fatalf("stopped Scan count = %d", stop.n)
	}
}

func TestGetPrefix(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	f.Put(srKey(1, 1, 0, 0, 0x0A))
	f.Put(srKey(1, 1, 0, 1, 0x0B))
	f.Put(srKey(1, 2, 0, 0, 0x0C))

	first, found := f.GetPrefix([]byte{1, 1})
	if !found {
		t.Fatal("GetPrefix 1,1 in memory")
	}
	if first[0] != 1 || first[1] != 1 {
		t.Fatal("GetPrefix returned a non-matching prefix")
	}
	second, found2 := f.GetPrefix([]byte{1, 2})
	if !found2 || second[2] != 0 {
		t.Fatal("GetPrefix 1,2")
	}
	if _, found3 := f.GetPrefix([]byte{9}); found3 {
		t.Fatal("GetPrefix of an absent prefix")
	}

	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	diskFirst, dfound := f.GetPrefix([]byte{1, 1})
	if !dfound || diskFirst[0] != 1 || diskFirst[1] != 1 {
		t.Fatal("GetPrefix from disk")
	}
}

func TestClearAndSkipFlush(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	f.Put(srRec(1, 0x01))
	f.Put(srRec(2, 0x02))
	if _, qerr := f.QuickFlush(3, false); qerr != nil {
		t.Fatal(qerr)
		return
	}
	if f.Count() != 2 {
		t.Fatalf("Count before Clear = %d", f.Count())
	}
	if err := f.Clear(); err != nil {
		t.Fatal(err)
		return
	}
	if f.Count() != 0 || f.Dirty() {
		t.Fatalf("Clear left count = %d", f.Count())
	}
	if f.Dirty() {
		t.Fatal("Clear left the file dirty")
	}
	if _, found := f.Get(srRec(1, 0)); found {
		t.Fatal("Get after Clear")
	}
	if _, lok := f.Last(); lok {
		t.Fatal("Last after Clear")
	}

	f.Put(srRec(4, 0x04))
	if !f.Dirty() {
		t.Fatal("Put after Clear must be dirty")
	}
	f.SkipFlush()
	if f.Dirty() {
		t.Fatal("SkipFlush must drop the pending buffer")
	}
}

func TestRecSorter(t *testing.T) {
	data := []byte{}
	data = data | srRec(3, 0x03)
	data = data | srRec(1, 0x01)
	data = data | srRec(2, 0x02)
	tmp := []byte{:tRecLen}
	rs := &recSorter{data: data, recLen: tRecLen, cmpLen: tCmpLen, tmp: tmp}
	if rs.Len() != 3 {
		t.Fatalf("recSorter Len = %d", rs.Len())
	}
	if !rs.Less(1, 0) {
		t.Fatal("recSorter Less")
	}
	sort.Sort(rs)
	if rs.data[0] != 1 || rs.data[8] != 2 || rs.data[16] != 3 {
		t.Fatal("recSorter sort order")
	}
	if rs.data[4] != 0x01 || rs.data[12] != 0x02 || rs.data[20] != 0x03 {
		t.Fatal("recSorter moved payloads with their keys")
	}
}

func TestOpenMissingDirectory(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)
	if _, err := Open(dir|"/nope/idx.dat", tRecLen, tCmpLen); err == nil {
		t.Fatal("Open in a missing directory must fail")
	}
}

// TestKeyByteOrder pins that ordering is byte-lexicographic on cmpLen bytes and
// ignores the trailing payload.
func TestKeyByteOrder(t *testing.T) {
	dir, ok := srTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	f, ok2 := srOpen(t, dir|"/idx.dat")
	if !ok2 {
		return
	}
	defer f.Close()

	f.Put(srKey(0x01, 0x00, 0x00, 0x00, 0xFF))
	f.Put(srKey(0x00, 0xFF, 0xFF, 0xFF, 0x00))
	if err := f.Flush(); err != nil {
		t.Fatal(err)
		return
	}
	sink := newSortedSink()
	f.Scan(srBound(0), srBound(255), sink.add)
	if sink.n != 2 {
		t.Fatalf("order Scan count = %d", sink.n)
	}
	if sink.recs[0][0] != 0x00 || sink.recs[1][0] != 0x01 {
		t.Fatal("byte-lexicographic order violated")
	}
	if !bytes.Equal(sink.recs[0], srKey(0x00, 0xFF, 0xFF, 0xFF, 0x00)) {
		t.Fatal("payload not carried with the record")
	}
}