package dbdoor

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"reflect"
	"strings"
	"testing"

	"github.com/benbjohnson/litestream"
	lsfile "github.com/benbjohnson/litestream/file"
	"github.com/superfly/ltx"

	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
)

func TestPlanRestoreSizesTheRestoreExactly(t *testing.T) {
	src := t.TempDir()
	dbdoortest.Replicate(t, src, 4)
	plan, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0)
	if err != nil {
		t.Fatal(err)
	}
	r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), 0)
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(r.Dir)
	st, err := os.Stat(r.Path)
	if err != nil {
		t.Fatal(err)
	}
	if plan.RestoredBytes != st.Size() || plan.TXID != r.TXID || plan.LTXBytes <= 0 || plan.Files == 0 || plan.Target != 0 {
		t.Fatalf("plan %+v, restored %d bytes at %d; want the plan's size and txid", plan, st.Size(), r.TXID)
	}
	// With no snapshot the plan is every L0 file, so LTXBytes is the
	// listing's sizes summed.
	var listed int64
	files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0)
	for _, f := range files {
		listed += f.Size
	}
	if plan.Files != len(files) || plan.LTXBytes != listed {
		t.Fatalf("plan %+v, want %d files of %d bytes", plan, len(files), listed)
	}
}

func TestRestoreToATargetTXID(t *testing.T) {
	src := t.TempDir()
	dbdoortest.Replicate(t, src, 4)
	files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0)
	at := uint64(files[0].MaxTXID)
	r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), at)
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(r.Dir)
	if r.TXID != at {
		t.Fatalf("restored to %d, want the first file's end %d", r.TXID, at)
	}
	if rows := countRows(t, r.Path); rows != dbdoortest.Rows {
		t.Fatalf("rows = %d, want one batch's %d", rows, dbdoortest.Rows)
	}
}

// A target at or past the latest transaction plans the latest copy, and
// says so with Target 0: litestream itself refuses a target past the end.
func TestPlanRestoreTargetPastTheEndPlansTheLatest(t *testing.T) {
	src := t.TempDir()
	dbdoortest.Replicate(t, src, 3)
	latest, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0)
	if err != nil {
		t.Fatal(err)
	}
	for _, target := range []uint64{latest.TXID, latest.TXID + 100} {
		got, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), target)
		if err != nil {
			t.Fatalf("target %d: %v", target, err)
		}
		if !reflect.DeepEqual(got, latest) {
			t.Fatalf("target %d planned %+v, want the latest %+v", target, got, latest)
		}
	}
}

// A planned TXID pins the restore: after the source moves on, a restore at
// plan.TXID still produces exactly the planned size, transaction and rows.
// Task 5 restores at plan.TXID for that reason (preflight ruling M2).
func TestRestorePinnedAtAPlannedTXIDKeepsThePlannedSize(t *testing.T) {
	dir := t.TempDir()
	db := dbdoortest.Replicate(t, dir, 2)
	plan, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0)
	if err != nil {
		t.Fatal(err)
	}

	// The source grows by many pages and replicates again.
	grow(t, db)
	dbdoortest.Sync(t, db, dir)
	later, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0)
	if err != nil {
		t.Fatal(err)
	}
	if later.TXID <= plan.TXID || later.RestoredBytes <= plan.RestoredBytes {
		t.Fatalf("the source did not move on: planned %+v, later %+v", plan, later)
	}

	r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), plan.TXID)
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(r.Dir)
	st, err := os.Stat(r.Path)
	if err != nil {
		t.Fatal(err)
	}
	if st.Size() != plan.RestoredBytes || r.TXID != plan.TXID {
		t.Fatalf("pinned restore: %d bytes at %d, want the plan's %d bytes at %d", st.Size(), r.TXID, plan.RestoredBytes, plan.TXID)
	}
	if rows := countRows(t, r.Path); rows != 2*dbdoortest.Rows {
		t.Fatalf("rows = %d, want the planned %d", rows, 2*dbdoortest.Rows)
	}
}

// Retention keeps a snapshot and the L0 files after it. A target inside
// the snapshot's span is history that is no longer kept, while the latest
// copy (and a target at the snapshot's own end) still restores.
func TestPlanRestoreTargetBeforeHistory(t *testing.T) {
	src := t.TempDir()
	dbdoortest.Replicate(t, src, 4)
	files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0)
	if len(files) < 4 {
		t.Fatalf("want one L0 file per batch, got %d", len(files))
	}
	snap := snapshotThrough(t, src, files[:2])
	for _, f := range files[:2] {
		if err := os.Remove(litestream.LTXFilePath(src, 0, f.MinTXID, f.MaxTXID)); err != nil {
			t.Fatal(err)
		}
	}

	latest, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0)
	if err != nil {
		t.Fatalf("the latest plan over a snapshot and later L0 files: %v", err)
	}
	if latest.TXID != uint64(files[len(files)-1].MaxTXID) || latest.Files != 1+len(files)-2 {
		t.Fatalf("latest plan %+v, want the snapshot plus %d L0 files ending at %d", latest, len(files)-2, files[len(files)-1].MaxTXID)
	}
	r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), latest.TXID)
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(r.Dir)
	if rows := countRows(t, r.Path); rows != 4*dbdoortest.Rows {
		t.Fatalf("rows = %d, want every batch's %d", rows, 4*dbdoortest.Rows)
	}
	if st, err := os.Stat(r.Path); err != nil || st.Size() != latest.RestoredBytes {
		t.Fatalf("restored over the snapshot: stat %v, err %v; want the planned %d bytes", st, err, latest.RestoredBytes)
	}

	// A target at an L0 file's end after the snapshot: the snapshot plus
	// that one file.
	mid := uint64(files[2].MaxTXID)
	atMid, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), mid)
	if err != nil || atMid.TXID != mid || atMid.Files != 2 || atMid.Target != mid {
		t.Fatalf("a target after the snapshot: plan %+v, err %v; want the snapshot and one L0 file ending at %d", atMid, err, mid)
	}
	rm, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), atMid.TXID)
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(rm.Dir)
	if st, err := os.Stat(rm.Path); err != nil || st.Size() != atMid.RestoredBytes || countRows(t, rm.Path) != 3*dbdoortest.Rows {
		t.Fatalf("restored at %d: stat %v, err %v; want %d bytes and three batches", mid, st, err, atMid.RestoredBytes)
	}

	atSnap, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), uint64(snap.MaxTXID))
	if err != nil || atSnap.TXID != uint64(snap.MaxTXID) || atSnap.Files != 1 || atSnap.Target != uint64(snap.MaxTXID) {
		t.Fatalf("a target at the snapshot's end: plan %+v, err %v; want the snapshot alone", atSnap, err)
	}

	_, err = PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), uint64(files[0].MaxTXID))
	if !errors.Is(err, ErrHistoryGone) || errors.Is(err, ErrNothingReplicated) {
		t.Fatalf("a target before the retained history: err = %v, want ErrHistoryGone", err)
	}
	_, err = RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), uint64(files[0].MaxTXID))
	if !errors.Is(err, ErrHistoryGone) || errors.Is(err, ErrNothingReplicated) {
		t.Fatalf("a restore before the retained history: err = %v, want ErrHistoryGone", err)
	}

	if _, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), 0); !errors.Is(err, ErrNothingReplicated) {
		t.Fatalf("an empty replica: err = %v, want ErrNothingReplicated", err)
	}
	if _, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), 5); !errors.Is(err, ErrNothingReplicated) {
		t.Fatalf("an empty replica with a target: err = %v, want ErrNothingReplicated", err)
	}
}

// snapshotThrough compacts the given L0 files, which must start at TXID 1,
// into one snapshot-level file in the file replica at dir, the way
// litestream's own compactor writes a level: ltx.Compactor over the source
// files, written through the file client.
func snapshotThrough(t *testing.T, dir string, files []*ltx.FileInfo) *ltx.FileInfo {
	t.Helper()
	ctx := context.Background()
	rc := lsfile.NewReplicaClient(dir)
	var rdrs []io.Reader
	for _, f := range files {
		r, err := rc.OpenLTXFile(ctx, 0, f.MinTXID, f.MaxTXID, 0, 0)
		if err != nil {
			t.Fatal(err)
		}
		defer r.Close()
		rdrs = append(rdrs, r)
	}
	pr, pw := io.Pipe()
	go func() {
		c, err := ltx.NewCompactor(pw, rdrs)
		if err != nil {
			pw.CloseWithError(err)
			return
		}
		c.HeaderFlags = ltx.HeaderFlagNoChecksum
		pw.CloseWithError(c.Compact(ctx))
	}()
	info, err := rc.WriteLTXFile(ctx, litestream.SnapshotLevel, files[0].MinTXID, files[len(files)-1].MaxTXID, pr)
	pr.CloseWithError(err)
	if err != nil {
		t.Fatal(err)
	}
	return info
}

// grow adds enough rows to the source database at path to add many pages.
func grow(t *testing.T, path string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	for i := 0; i < 50; i++ {
		if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("grow-%d-%s", i, strings.Repeat("x", 2000))); err != nil {
			t.Fatal(err)
		}
	}
}

// The header PlanRestore sizes the restore from comes from the replica,
// so it is untrusted. A page size SQLite cannot have, a zero commit, or
// TXIDs that are not the listed file's are refused rather than sized; an
// unchecked Commit times PageSize can wrap negative and slip under any
// size cap.
func TestPlanRestoreRefusesAHeaderThatDoesNotDescribeItsFile(t *testing.T) {
	src := t.TempDir()
	dbdoortest.Replicate(t, src, 2)
	files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0)
	last := files[len(files)-1]
	rc, err := lsfile.NewReplicaClient(src).OpenLTXFile(context.Background(), 0, last.MinTXID, last.MaxTXID, 0, ltx.HeaderSize)
	if err != nil {
		t.Fatal(err)
	}
	honest := make([]byte, ltx.HeaderSize)
	_, err = io.ReadFull(rc, honest)
	rc.Close()
	if err != nil {
		t.Fatal(err)
	}

	for _, tc := range []struct {
		name string
		edit func(h *ltx.Header)
	}{
		{"a page size that wraps the size", func(h *ltx.Header) { h.PageSize, h.Commit = 0xFFFFFFFF, 0xFFFFFFFF }},
		{"a page size that is not a power of two", func(h *ltx.Header) { h.PageSize = 4000 }},
		{"a zero page size", func(h *ltx.Header) { h.PageSize = 0 }},
		{"a zero commit", func(h *ltx.Header) { h.Commit = 0 }},
		{"another file's max txid", func(h *ltx.Header) { h.MaxTXID++ }},
		{"another file's min txid", func(h *ltx.Header) { h.MinTXID++ }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			var h ltx.Header
			if err := h.UnmarshalBinary(honest); err != nil {
				t.Fatal(err)
			}
			tc.edit(&h)
			crafted, _ := h.MarshalBinary()
			door := newTestDoor(t, src)
			headerTail := GetTail(0, uint64(last.MinTXID), uint64(last.MaxTXID))
			do := func(ctx context.Context, tail string, hd http.Header) (*http.Response, error) {
				if tail == headerTail && hd.Get("Range") == RangeHeader(0, ltx.HeaderSize) {
					rec := httptest.NewRecorder()
					rec.WriteHeader(http.StatusPartialContent)
					rec.Write(crafted)
					return rec.Result(), nil
				}
				return door.do(ctx, tail, hd)
			}
			plan, err := PlanRestore(context.Background(), NewClient(do), 0)
			if err == nil || errors.Is(err, ErrNothingReplicated) || errors.Is(err, ErrHistoryGone) {
				t.Fatalf("plan %+v, err = %v; want the header refused as its own error", plan, err)
			}
			if !reflect.DeepEqual(plan, RestorePlan{}) {
				t.Fatalf("a refused header still planned %+v", plan)
			}
		})
	}
}

// A database that shrinks (rows deleted, then VACUUM) is sized by its last
// file's commit, not by the largest the history ever held.
func TestPlanRestoreFollowsAShrinkingDatabase(t *testing.T) {
	dir := t.TempDir()
	db := dbdoortest.Replicate(t, dir, 1)
	grow(t, db)
	dbdoortest.Sync(t, db, dir)
	big, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0)
	if err != nil {
		t.Fatal(err)
	}
	sdb, err := sql.Open("sqlite", db)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := sdb.Exec(`DELETE FROM t WHERE body LIKE 'grow-%'; VACUUM`); err != nil {
		sdb.Close()
		t.Fatal(err)
	}
	sdb.Close()
	dbdoortest.Sync(t, db, dir)

	small, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0)
	if err != nil {
		t.Fatal(err)
	}
	if small.TXID <= big.TXID || small.RestoredBytes >= big.RestoredBytes {
		t.Fatalf("after the shrink: planned %+v, before %+v; want a later, smaller plan", small, big)
	}
	r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), small.TXID)
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(r.Dir)
	st, err := os.Stat(r.Path)
	if err != nil {
		t.Fatal(err)
	}
	if st.Size() != small.RestoredBytes {
		t.Fatalf("restored %d bytes, planned %d", st.Size(), small.RestoredBytes)
	}
}