package wal

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

type walSink struct {
	serials []uint64
	data    [][]byte
	n       int32
}

func newWalSink() (s *walSink) {
	s = &walSink{
		serials: []uint64{:32},
		data:    [][]byte{:32},
	}
	return
}

func (s *walSink) add(ser uint64, data []byte) (ok bool) {
	if s.n >= int32(len(s.serials)) {
		return false
	}
	cp := []byte{:len(data)}
	copy(cp, data)
	s.serials[s.n] = ser
	s.data[s.n] = cp
	s.n++
	return true
}

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

// TestAppendReadRoundTrip pins the serial encoding (segment<<32 | offset) and
// the length-prefixed entry format that every reader depends on.
func TestAppendReadRoundTrip(t *testing.T) {
	dir, ok := walTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer w.Close()

	e1 := []byte("alpha")
	e2 := []byte("beta")
	e3 := []byte{:1000}
	for i := range e3 {
		e3[i] = byte(i)
	}

	s1, err1 := w.Append(e1)
	if err1 != nil {
		t.Fatal(err1)
		return
	}
	s2, err2 := w.Append(e2)
	if err2 != nil {
		t.Fatal(err2)
		return
	}
	s3, err3 := w.Append(e3)
	if err3 != nil {
		t.Fatal(err3)
		return
	}

	// Segment 0, first byte at offset 0.
	if s1 != 0 {
		t.Fatalf("first serial = %d", s1)
	}
	if s2 != uint64(HdrSize+len(e1)) {
		t.Fatalf("second serial = %d", s2)
	}
	if s3 != uint64(HdrSize+len(e1)+HdrSize+len(e2)) {
		t.Fatalf("third serial = %d", s3)
	}
	if w.cur != 0 {
		t.Fatalf("cur = %d", w.cur)
	}
	if w.off != int64(HdrSize+len(e1)+HdrSize+len(e2)+HdrSize+len(e3)) {
		t.Fatalf("off = %d", w.off)
	}

	g1, gerr1 := w.Read(s1)
	if gerr1 != nil {
		t.Fatal(gerr1)
		return
	}
	if !bytes.Equal(g1, e1) {
		t.Fatal("entry 1 round trip")
	}
	g2, gerr2 := w.Read(s2)
	if gerr2 != nil {
		t.Fatal(gerr2)
		return
	}
	if !bytes.Equal(g2, e2) {
		t.Fatal("entry 2 round trip")
	}
	g3, gerr3 := w.Read(s3)
	if gerr3 != nil {
		t.Fatal(gerr3)
		return
	}
	if !bytes.Equal(g3, e3) {
		t.Fatal("entry 3 round trip")
	}
}

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

	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer w.Close()

	// Empty segment: offset 0 has no header.
	if _, e1 := w.Read(0); e1 == nil {
		t.Fatal("read from an empty segment must fail")
	}
	// Segment index past the end.
	if _, e2 := w.Read(uint64(1) << 32); e2 == nil {
		t.Fatal("read from a missing segment must fail")
	}
	// Negative segment index after the int32 cast.
	if _, e3 := w.Read(uint64(0xFFFFFFFFFFFFFFFF)); e3 == nil {
		t.Fatal("read from a negative segment must fail")
	}
}

// TestReadImpossibleLength plants a length prefix larger than MaxEntrySize and
// checks the guard rejects it instead of allocating the value.
func TestReadImpossibleLength(t *testing.T) {
	dir, ok := walTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	bad := []byte{0xFF, 0xFF, 0xFF, 0xFF}
	if werr := os.WriteFile(dir|"/seg-000.dat", bad, 0644); werr != nil {
		t.Fatal(werr)
		return
	}
	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer w.Close()

	if _, rerr := w.Read(0); rerr == nil {
		t.Fatal("an impossible length must fail the read")
	}
	sink := newWalSink()
	if ferr := w.ForEach(sink.add); ferr != nil {
		t.Fatal(ferr)
		return
	}
	if sink.n != 0 {
		t.Fatalf("ForEach yielded %d entries from a bad header", sink.n)
	}
}

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

	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer w.Close()

	names := [][]byte{[]byte("one"), []byte("two"), []byte("three")}
	for _, n := range names {
		if _, aerr := w.Append(n); aerr != nil {
			t.Fatal(aerr)
			return
		}
	}

	sink := newWalSink()
	if ferr := w.ForEach(sink.add); ferr != nil {
		t.Fatal(ferr)
		return
	}
	if sink.n != 3 {
		t.Fatalf("ForEach count = %d", sink.n)
	}
	for i := 0; i < 3; i++ {
		if !bytes.Equal(sink.data[i], names[i]) {
			t.Fatalf("entry %d = %s", i, string(sink.data[i]))
		}
	}
	if !(sink.serials[0] < sink.serials[1] && sink.serials[1] < sink.serials[2]) {
		t.Fatal("ForEach serials not ascending")
	}
}

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

	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer w.Close()

	a := []byte("aaa")
	b := []byte("bbb")
	c := []byte("ccc")
	sa, ea := w.Append(a)
	if ea != nil {
		t.Fatal(ea)
		return
	}
	sb, eb := w.Append(b)
	if eb != nil {
		t.Fatal(eb)
		return
	}
	sc, ec := w.Append(c)
	if ec != nil {
		t.Fatal(ec)
		return
	}

	// startSer == 0 iterates everything.
	sink := newWalSink()
	if ferr := w.ForEachFrom(0, sink.add); ferr != nil {
		t.Fatal(ferr)
		return
	}
	if sink.n != 3 {
		t.Fatalf("ForEachFrom(0) count = %d", sink.n)
	}

	// startSer is exclusive: from b, only c follows.
	sink2 := newWalSink()
	if ferr2 := w.ForEachFrom(sb, sink2.add); ferr2 != nil {
		t.Fatal(ferr2)
		return
	}
	if sink2.n != 1 || !bytes.Equal(sink2.data[0], c) {
		t.Fatalf("ForEachFrom(b) count = %d", sink2.n)
	}
	if sink2.serials[0] != sc {
		t.Fatal("ForEachFrom(b) serial")
	}

	// From the last entry: nothing follows.
	sink3 := newWalSink()
	if ferr3 := w.ForEachFrom(sc, sink3.add); ferr3 != nil {
		t.Fatal(ferr3)
		return
	}
	if sink3.n != 0 {
		t.Fatalf("ForEachFrom(c) count = %d", sink3.n)
	}

	// Serial 0 is the first entry's serial and also the "iterate all"
	// sentinel: ForEachFrom(sa) must behave like ForEachFrom(0) and include a.
	if sa != 0 {
		t.Fatalf("first serial = %d, expected the serial-0 sentinel", sa)
	}
	sink4 := newWalSink()
	if ferr4 := w.ForEachFrom(sa, sink4.add); ferr4 != nil {
		t.Fatal(ferr4)
		return
	}
	if sink4.n != 3 || !bytes.Equal(sink4.data[0], a) {
		t.Fatalf("ForEachFrom(0 sentinel) count = %d", sink4.n)
	}
}

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

	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer w.Close()

	if _, aerr := w.Append([]byte("payload")); aerr != nil {
		t.Fatal(aerr)
		return
	}

	sink := newWalSink()
	// Offset well past the segment size means the checkpoint refers to a
	// segment that was truncated: callers must see ErrCheckpointStale.
	if ferr := w.ForEachFrom(1000, sink.add); ferr != ErrCheckpointStale {
		t.Fatalf("stale offset error = %v", ferr)
	}
	// A segment index that does not exist yet is not stale, just empty.
	sink2 := newWalSink()
	if ferr2 := w.ForEachFrom(uint64(5)<<32, sink2.add); ferr2 != nil {
		t.Fatal(ferr2)
		return
	}
	if sink2.n != 0 {
		t.Fatalf("missing segment yielded %d entries", sink2.n)
	}
	// An offset exactly at end of file skips nothing and yields nothing.
	sink3 := newWalSink()
	if ferr3 := w.ForEachFrom(w.off, sink3.add); ferr3 != nil {
		t.Fatal(ferr3)
		return
	}
	if sink3.n != 0 {
		t.Fatalf("end-of-file offset yielded %d entries", sink3.n)
	}
}

// TestForEachTruncatedEntry cuts the payload off the second entry and checks
// ForEach stops at the torn record instead of reading past it.
func TestForEachTruncatedEntry(t *testing.T) {
	dir, ok := walTmp(t)
	if !ok {
		return
	}
	defer os.RemoveAll(dir)

	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}

	first := []byte("first-payload")
	second := []byte("second-payload")
	if _, aerr := w.Append(first); aerr != nil {
		t.Fatal(aerr)
		return
	}
	full := w.off
	if _, berr := w.Append(second); berr != nil {
		t.Fatal(berr)
		return
	}
	if terr := w.segs[0].Truncate(full + int64(HdrSize) + 2); terr != nil {
		t.Fatal(terr)
		return
	}
	w.Close()

	// Reopen so the segment size is re-stat'ed from the truncated file.
	w2, err2 := Open(dir)
	if err2 != nil {
		t.Fatal(err2)
		return
	}
	defer w2.Close()

	sink := newWalSink()
	if ferr := w2.ForEach(sink.add); ferr != nil {
		t.Fatal(ferr)
		return
	}
	if sink.n != 1 || !bytes.Equal(sink.data[0], first) {
		t.Fatalf("truncated ForEach count = %d", sink.n)
	}
}

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

	w1, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	a := []byte("entry-a")
	sa, aerr := w1.Append(a)
	if aerr != nil {
		t.Fatal(aerr)
		return
	}
	if serr := w1.Sync(); serr != nil {
		t.Fatal(serr)
		return
	}
	if cerr := w1.Close(); cerr != nil {
		t.Fatal(cerr)
		return
	}

	w2, err2 := Open(dir)
	if err2 != nil {
		t.Fatal(err2)
		return
	}
	if len(w2.segs) != 1 {
		t.Fatalf("reopened segment count = %d", int32(len(w2.segs)))
	}
	if w2.off != int64(HdrSize+len(a)) {
		t.Fatalf("reopened off = %d", w2.off)
	}

	sink := newWalSink()
	if ferr := w2.ForEach(sink.add); ferr != nil {
		t.Fatal(ferr)
		return
	}
	if sink.n != 1 || !bytes.Equal(sink.data[0], a) {
		t.Fatalf("reopened ForEach count = %d", sink.n)
	}

	// The next append continues where the old one stopped, not over it.
	b := []byte("entry-bb")
	sb, berr := w2.Append(b)
	if berr != nil {
		t.Fatal(berr)
		return
	}
	if sb != sa+uint64(HdrSize+len(a)) {
		t.Fatalf("append after reopen serial = %d", sb)
	}
	gb, gerr := w2.Read(sb)
	if gerr != nil {
		t.Fatal(gerr)
		return
	}
	if !bytes.Equal(gb, b) {
		t.Fatal("append after reopen round trip")
	}
	if cerr2 := w2.Close(); cerr2 != nil {
		t.Fatal(cerr2)
		return
	}
}

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

	w, err := Open(dir)
	if err != nil {
		t.Fatal(err)
		return
	}
	defer w.Close()

	if len(w.segs) != 1 {
		t.Fatalf("fresh WAL segment count = %d", int32(len(w.segs)))
	}
	if w.off != 0 {
		t.Fatalf("fresh WAL off = %d", w.off)
	}
	sink := newWalSink()
	if ferr := w.ForEach(sink.add); ferr != nil {
		t.Fatal(ferr)
		return
	}
	if sink.n != 0 {
		t.Fatalf("empty WAL yielded %d entries", sink.n)
	}
}