package hostbox

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/superfly/ltx"
	_ "modernc.org/sqlite"
)

type restoreReader struct {
	data       map[string][]byte
	openedKey  string
	openedVers string
}

func (*restoreReader) ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error) {
	return nil, errors.New("unexpected listing")
}
func (*restoreReader) GetVersion(context.Context, string, string, string) ([]byte, error) {
	return nil, errors.New("unexpected get")
}
func (r *restoreReader) OpenVersion(_ context.Context, _ string, key, version string) (io.ReadCloser, error) {
	r.openedKey, r.openedVers = key, version
	b, ok := r.data[key+"@"+version]
	if !ok {
		return nil, errors.New("missing immutable version")
	}
	return io.NopCloser(bytes.NewReader(b)), nil
}

func TestLoadPinnedReadsTheSelectedVersionAndRejectsCorruption(t *testing.T) {
	r := &restoreReader{data: map[string][]byte{
		"instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx@selected": []byte("corrupt selected object"),
		"instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx@current":  []byte("different current object"),
	}}
	p := &genselect.Plan{Generation: "one", MaxTXID: 1, Files: []genselect.File{{
		Key: "instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx", VersionID: "selected", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len("corrupt selected object")),
	}}}
	if _, _, err := loadPinned(context.Background(), r, "bucket", p); err == nil {
		t.Fatal("loadPinned accepted an object without a valid complete LTX trailer")
	}
	if r.openedVers != "selected" {
		t.Fatalf("opened version %q, want immutable selected version", r.openedVers)
	}
}

type restoreWork struct {
	calls  int
	active bool
}

func (w *restoreWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
	w.calls++
	w.active = true
	defer func() { w.active = false }()
	return fn()
}

func TestColdRestoreRefusesBeforeMutationWithoutSelectedAuthority(t *testing.T) {
	r := ColdRestore{}
	if err := r.Install(context.Background(), t.TempDir()+"/db", &genselect.Plan{}); !errors.Is(err, ErrUnsupported) {
		t.Fatalf("Install error = %v, want ErrUnsupported", err)
	}
}

func TestColdRestorePurgesSameGenerationUnderSelectedWork(t *testing.T) {
	dir := t.TempDir()
	db := filepath.Join(dir, "shop.db")
	if err := os.WriteFile(db, []byte("warm database must not be reused"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(db+".gen", []byte("one\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	key, snapshot, crc := snapshotFixture(t, "selected")
	rd := &restoreReader{data: map[string][]byte{key + "@v1": snapshot}}
	work := &restoreWork{}
	err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "one", Work: work}).Install(context.Background(), db,
		&genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: crc}, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(snapshot))}}})
	if err != nil {
		t.Fatalf("Install: %v", err)
	}
	if work.calls != 2 {
		t.Fatalf("selected work calls = %d, want capture and install", work.calls)
	}
	dbcheck, err := sql.Open("sqlite", db)
	if err != nil {
		t.Fatal(err)
	}
	defer dbcheck.Close()
	var got string
	if err := dbcheck.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "selected" {
		t.Fatalf("restored value=%q err=%v", got, err)
	}
	if string(mustRead(t, db+".gen")) != "one\n" {
		t.Fatal("generation marker not installed")
	}
}

func TestColdRestorePinsVersionAndRetiresForeignGenerationOutsideWork(t *testing.T) {
	dir := t.TempDir()
	db := filepath.Join(dir, "shop.db")
	if err := os.WriteFile(db, []byte("foreign"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(db+".gen", []byte("old\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	key, v1, crc := snapshotFixture(t, "selected-v1")
	_, v2, _ := snapshotFixture(t, "current-v2")
	rd := &restoreReader{data: map[string][]byte{key + "@v1": v1, key + "@v2": v2}}
	work := &restoreWork{}
	retired := ""
	parked := filepath.Join(dir, "parked")
	err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "new", Work: work, RetireArtifacts: func(_ context.Context, g string) error {
		if work.active {
			t.Fatal("retirement ran under Work flock")
		}
		retired = g
		if err := os.Mkdir(parked, 0o700); err != nil {
			return err
		}
		if err := os.Rename(db, filepath.Join(parked, "shop.db")); err != nil {
			return err
		}
		return os.Rename(db+".gen", filepath.Join(parked, "shop.db.gen"))
	}}).Install(context.Background(), db,
		&genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: crc}, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(v1))}}})
	if err != nil {
		t.Fatal(err)
	}
	if retired != "old" {
		t.Fatalf("retired %q, want old", retired)
	}
	if rd.openedVers != "v1" {
		t.Fatalf("opened %q, want pinned v1", rd.openedVers)
	}
	check, err := sql.Open("sqlite", db)
	if err != nil {
		t.Fatal(err)
	}
	defer check.Close()
	var got string
	if err := check.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "selected-v1" {
		t.Fatalf("restored value=%q err=%v", got, err)
	}
	if string(mustRead(t, db+".gen")) != "new\n" {
		t.Fatalf("installed marker=%q, want admitted generation", mustRead(t, db+".gen"))
	}
}

func TestColdRestoreReportsPurgeFailureBeforeInstall(t *testing.T) {
	dir := t.TempDir()
	db := filepath.Join(dir, "shop.db")
	if err := os.Mkdir(db, 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(db, "still-here"), []byte("x"), 0o600); err != nil {
		t.Fatal(err)
	}
	key, snapshot, _ := snapshotFixture(t, "never-installed")
	rd := &restoreReader{data: map[string][]byte{key + "@v1": snapshot}}
	err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "one", Work: &restoreWork{}}).Install(context.Background(), db,
		&genselect.Plan{Generation: "one", MaxTXID: 1, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(snapshot))}}})
	if err == nil || !strings.Contains(err.Error(), "purge "+db) {
		t.Fatalf("Install error=%v, want checked purge failure", err)
	}
	if _, err := os.Stat(filepath.Join(db, "still-here")); err != nil {
		t.Fatalf("purge failure damaged existing artifact: %v", err)
	}
}

func TestColdRestoreInstallEmptyRequiresProofAndWritesOnlyGeneration(t *testing.T) {
	dir := t.TempDir()
	db := filepath.Join(dir, "empty.db")
	work := &restoreWork{}
	proved := false
	r := ColdRestore{Generation: "new", Work: work}
	if err := r.InstallEmpty(context.Background(), db, func(context.Context) error { proved = true; return nil }); err != nil {
		t.Fatal(err)
	}
	if !proved || work.calls != 2 {
		t.Fatalf("proved=%v guarded calls=%d", proved, work.calls)
	}
	if _, err := os.Stat(db); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("empty bootstrap wrote DB: %v", err)
	}
	if string(mustRead(t, db+".gen")) != "new\n" {
		t.Fatal("admitted generation missing")
	}
}

func TestColdRestoreAcceptsSelectorOverlappingOutput(t *testing.T) {
	for _, min := range []ltx.TXID{10, 12} {
		t.Run(fmt.Sprintf("continuation-%d-13", min), func(t *testing.T) {
			ctx := t.Context()
			bucket := leasetest.NewVersionedBucket()
			bucket.AddBucket("data")
			bucket.PageSize = 1
			key, snapshot, crc := restoreLTXFixture(t, "snapshot-at-11", 9, 1, 11)
			version := bucket.Put("data", key, snapshot)
			continuation, data, _ := restoreLTXFixture(t, "compacted-at-13", 1, min, 13)
			bucket.Put("data", continuation, data)
			plan, err := genselect.SelectPrepared(ctx, bucket, genselect.Request{
				Bucket: "data", Region: "ie", Host: "shop",
				Holder: hostclaim.Holder{Incarnation: "inc", Location: "instances/shop"},
			}, genselect.PreparedPin{Generation: "one", SnapshotKey: key, SnapshotVersion: version, SnapshotTXID: 11, SnapshotCRC64: crc})
			if err != nil {
				t.Fatal(err)
			}
			if len(plan.Files) != 2 || plan.Files[0].MinTXID != 1 || plan.Files[0].MaxTXID != 11 || plan.Files[1].MinTXID != uint64(min) || plan.Files[1].MaxTXID != 13 || plan.MaxTXID != 13 {
				t.Fatalf("selector output = %+v", plan)
			}
			path := filepath.Join(t.TempDir(), "restored.db")
			if err := (ColdRestore{Reader: bucket, Bucket: "data", Generation: "new", Work: &restoreWork{}}).Install(ctx, path, plan); err != nil {
				t.Fatalf("restore selector output: %v", err)
			}
			db, err := sql.Open("sqlite", path)
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			var got string
			if err := db.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "compacted-at-13" {
				t.Fatalf("restored value = %q, err = %v", got, err)
			}
			if string(mustRead(t, path+".gen")) != "new\n" {
				t.Fatal("admitted generation marker missing")
			}
		})
	}
}

func TestLoadPinnedRejectsGapAndNonadvancingRange(t *testing.T) {
	for _, bounds := range [][2]ltx.TXID{{13, 13}, {10, 11}} {
		t.Run(fmt.Sprint(bounds), func(t *testing.T) {
			key, snapshot, _ := restoreLTXFixture(t, "snapshot", 9, 1, 11)
			continuation, data, _ := restoreLTXFixture(t, "invalid", 1, bounds[0], bounds[1])
			reader := &restoreReader{data: map[string][]byte{key + "@v1": snapshot, continuation + "@v2": data}}
			plan := &genselect.Plan{Generation: "one", MaxTXID: uint64(bounds[1]), Files: []genselect.File{
				{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 11, Size: int64(len(snapshot))},
				{Key: continuation, VersionID: "v2", Level: 1, MinTXID: uint64(bounds[0]), MaxTXID: uint64(bounds[1]), Size: int64(len(data))},
			}}
			if _, _, err := loadPinned(t.Context(), reader, "data", plan); !errors.Is(err, genselect.ErrGap) {
				t.Fatalf("invalid continuation accepted: %v", err)
			}
		})
	}
}

func snapshotFixture(t *testing.T, value string) (string, []byte, uint64) {
	t.Helper()
	return restoreLTXFixture(t, value, 9, 1, 1)
}

func restoreLTXFixture(t *testing.T, value string, level int, min, max ltx.TXID) (string, []byte, uint64) {
	t.Helper()
	path := filepath.Join(t.TempDir(), "fixture.db")
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = db.Exec(`CREATE TABLE fixture(value TEXT); INSERT INTO fixture VALUES (?)`, value); err != nil {
		t.Fatal(err)
	}
	if err = db.Close(); err != nil {
		t.Fatal(err)
	}
	return ltxFromDBFile(t, path, level, min, max)
}

// ltxFromDBFile cuts the database file at path into one LTX object and
// returns its key, bytes and file checksum.
func ltxFromDBFile(t *testing.T, path string, level int, min, max ltx.TXID) (string, []byte, uint64) {
	t.Helper()
	raw := mustRead(t, path)
	pageSize := int(binary.BigEndian.Uint16(raw[16:18]))
	if pageSize == 1 {
		pageSize = 65536
	}
	if len(raw)%pageSize != 0 {
		t.Fatalf("database size %d not page aligned", len(raw))
	}
	spec := ltx.FileSpec{Header: ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: uint32(pageSize), Commit: uint32(len(raw) / pageSize), MinTXID: min, MaxTXID: max}}
	for off := 0; off < len(raw); off += pageSize {
		pgno := uint32(off/pageSize + 1)
		if pgno == ltx.LockPgno(uint32(pageSize)) {
			continue
		}
		spec.Pages = append(spec.Pages, ltx.PageSpec{Header: ltx.PageHeader{Pgno: pgno}, Data: append([]byte(nil), raw[off:off+pageSize]...)})
	}
	var out bytes.Buffer
	if _, err := spec.WriteTo(&out); err != nil {
		t.Fatal(err)
	}
	return fmt.Sprintf("instances/shop/g-one/%04x/%s", level, ltx.FormatFilename(min, max)), out.Bytes(), uint64(spec.Trailer.FileChecksum)
}

func mustRead(t *testing.T, path string) []byte {
	t.Helper()
	b, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	return b
}

// The landing writes <dbPath>.gen through the one marker writer this
// package's cold restore reads (moves Plan 2 ruling G2): the prepared
// generation's name as plain text, so a later admission's readGeneration
// accepts it. A name the restore would refuse is refused on the way in.
func TestGenerationMarkerIsSharedWithTheColdRestore(t *testing.T) {
	path := filepath.Join(t.TempDir(), "h.db.gen")
	if err := WriteGenerationMarker(path, "mv_abc"); err != nil {
		t.Fatal(err)
	}
	for _, read := range []func(string) (string, error){ReadGenerationMarker, readGeneration} {
		if got, err := read(path); err != nil || got != "mv_abc" {
			t.Fatalf("read back %q, %v", got, err)
		}
	}
	if got, err := ReadGenerationMarker(filepath.Join(t.TempDir(), "absent.gen")); err != nil || got != "" {
		t.Fatalf("absent: %q, %v", got, err)
	}
	for _, bad := range []string{"", ".", "..", "instances/h", "a\nb"} {
		if err := WriteGenerationMarker(path, bad); err == nil {
			t.Fatalf("%q was written", bad)
		}
	}
	if got, _ := ReadGenerationMarker(path); got != "mv_abc" {
		t.Fatalf("a refused write changed the marker: %q", got)
	}
}

func TestLoadPinnedNormalizesOnlyAfterOriginalProof(t *testing.T) {
	key, original, _ := snapshotFixture(t, "checksummed")
	var spec ltx.FileSpec
	if _, err := spec.ReadFrom(bytes.NewReader(original)); err != nil {
		t.Fatal(err)
	}
	checksum := ltx.Checksum(0)
	for _, page := range spec.Pages {
		checksum ^= ltx.ChecksumPage(page.Header.Pgno, page.Data)
	}
	checksum |= ltx.ChecksumFlag
	spec.Header.Flags = 0
	spec.Trailer.PostApplyChecksum = checksum
	var encoded bytes.Buffer
	if _, err := spec.WriteTo(&encoded); err != nil {
		t.Fatal(err)
	}
	original = append([]byte(nil), encoded.Bytes()...)
	for _, fault := range []string{"valid", "pin", "file-checksum", "database-checksum"} {
		t.Run(fault, func(t *testing.T) {
			body := append([]byte(nil), original...)
			pin := uint64(spec.Trailer.FileChecksum)
			switch fault {
			case "pin":
				pin ^= 1
			case "file-checksum":
				body[len(body)-1] ^= 1
			case "database-checksum":
				bad := spec
				bad.Trailer.PostApplyChecksum ^= 1
				var out bytes.Buffer
				if _, err := bad.WriteTo(&out); err != nil {
					t.Fatal(err)
				}
				body, pin = out.Bytes(), uint64(bad.Trailer.FileChecksum)
			}
			reader := &restoreReader{data: map[string][]byte{key + "@selected": body}}
			plan := &genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: pin}, Files: []genselect.File{{Key: key, VersionID: "selected", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(body))}}}
			files, final, err := loadPinned(t.Context(), reader, "data", plan)
			if fault != "valid" {
				if err == nil {
					t.Fatal("normalization bypassed original proof")
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if final != checksum || len(files) != 1 || !bytes.Equal(reader.data[key+"@selected"], original) {
				t.Fatal("normalization changed selected bytes or lost database checksum")
			}
			var converted ltx.FileSpec
			if _, err := converted.ReadFrom(bytes.NewReader(files[0].data)); err != nil {
				t.Fatal(err)
			}
			if !converted.Header.NoChecksum() || converted.Header.PreApplyChecksum != 0 || converted.Trailer.PostApplyChecksum != 0 {
				t.Fatal("incompatible restore stream")
			}
			if len(converted.Pages) != len(spec.Pages) {
				t.Fatal("page loss")
			}
			for i := range converted.Pages {
				if converted.Pages[i].Header != spec.Pages[i].Header || !bytes.Equal(converted.Pages[i].Data, spec.Pages[i].Data) {
					t.Fatal("normalization changed database pages")
				}
			}
			path := filepath.Join(t.TempDir(), "restored.db")
			if err := (ColdRestore{Reader: reader, Bucket: "data", Generation: "successor", Work: &restoreWork{}}).Install(t.Context(), path, plan); err != nil {
				t.Fatal(err)
			}
		})
	}
}