package lease

import (
	"bytes"
	"context"
	"errors"
	"io"
	"log/slog"
	"path/filepath"
	"testing"
	"time"

	"github.com/benbjohnson/litestream"
	lss3 "github.com/benbjohnson/litestream/s3"
	"github.com/superfly/ltx"
)

type generationFake struct {
	calls int
	ctx   context.Context
	files []*ltx.FileInfo
}

func (f *generationFake) Type() string                   { return "fake" }
func (f *generationFake) SetLogger(*slog.Logger)         {}
func (f *generationFake) Init(ctx context.Context) error { f.calls++; f.ctx = ctx; return nil }
func (f *generationFake) LTXFiles(ctx context.Context, _ int, _ ltx.TXID, _ bool) (ltx.FileIterator, error) {
	f.calls++
	f.ctx = ctx
	return ltx.NewFileInfoSliceIterator(f.files), nil
}
func (f *generationFake) OpenLTXFile(ctx context.Context, _ int, _ ltx.TXID, _ ltx.TXID, _ int64, _ int64) (io.ReadCloser, error) {
	f.calls++
	f.ctx = ctx
	return io.NopCloser(bytes.NewReader(nil)), nil
}
func (f *generationFake) WriteLTXFile(ctx context.Context, _ int, min, max ltx.TXID, _ io.Reader) (*ltx.FileInfo, error) {
	f.calls++
	f.ctx = ctx
	return &ltx.FileInfo{MinTXID: min, MaxTXID: max}, nil
}
func (f *generationFake) DeleteLTXFiles(ctx context.Context, _ []*ltx.FileInfo) error {
	f.calls++
	f.ctx = ctx
	return nil
}
func (f *generationFake) DeleteAll(context.Context) error { f.calls++; return nil }

func TestBindGenerationIsImmutableAndCanonical(t *testing.T) {
	leaser := lss3.NewLeaser()
	leaser.Path = "instances/shop.example/inc"
	f := &generationFake{}
	h := &Held{leaser: leaser, replicaClient: f}
	check := func(context.Context) error { return nil }
	if err := h.BindGeneration("instances/shop.example/inc/g-generation1", check); err != nil {
		t.Fatal(err)
	}
	if err := h.BindGeneration("instances/shop.example/inc/g-generation2", check); err != ErrGenerationBinding {
		t.Fatalf("rebind=%v", err)
	}
	for _, bad := range []string{"instances/shop.example/inc/g-a/child", "instances/shop.example/inc/x/../g-a", "instances/shop.example/other/g-a", "instances/shop.example/inc/g-"} {
		h2 := &Held{leaser: leaser, replicaClient: &generationFake{}}
		if err := h2.BindGeneration(bad, check); err != ErrGenerationBinding {
			t.Errorf("BindGeneration(%q)=%v", bad, err)
		}
	}
}

func TestGenerationClientValidatesBeforeStorage(t *testing.T) {
	f := &generationFake{}
	c := &generationClient{client: f, check: func(context.Context) error { return nil }}
	if _, err := c.WriteLTXFile(t.Context(), -1, 1, 1, bytes.NewReader(nil)); err != ErrGenerationBinding {
		t.Fatal(err)
	}
	if _, err := c.OpenLTXFile(t.Context(), 0, 2, 1, 0, 0); err != ErrGenerationBinding {
		t.Fatal(err)
	}
	if err := c.DeleteLTXFiles(t.Context(), []*ltx.FileInfo{nil}); err != ErrGenerationBinding {
		t.Fatal(err)
	}
	if err := c.DeleteAll(t.Context()); err == nil {
		t.Fatal("DeleteAll accepted")
	}
	if f.calls != 0 {
		t.Fatalf("invalid requests reached storage: %d", f.calls)
	}
}

func TestGenerationClientBoundsAndRetainsStreamContext(t *testing.T) {
	checks := 0
	f := &generationFake{}
	c := &generationClient{client: f, check: func(context.Context) error { checks++; return nil }}
	r, err := c.OpenLTXFile(t.Context(), litestream.SnapshotLevel, 1, 1, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	deadline, ok := f.ctx.Deadline()
	if !ok || time.Until(deadline) > generationRequestLimit || time.Until(deadline) < generationRequestLimit-time.Second {
		t.Fatalf("deadline=%v ok=%v", deadline, ok)
	}
	select {
	case <-f.ctx.Done():
		t.Fatal("stream context canceled before Close")
	default:
	}
	if err := r.Close(); err != nil {
		t.Fatal(err)
	}
	select {
	case <-f.ctx.Done():
	case <-time.After(time.Second):
		t.Fatal("stream Close did not cancel request context")
	}
	if checks != 1 {
		t.Fatalf("authority checks=%d", checks)
	}
}

// GenerationTestBinding is a test-only bridge for the external package to
// exercise hostbox.GenerationPrefix without a lease -> hostbox import cycle.
func GenerationTestBinding(location, prefix string) error {
	l := lss3.NewLeaser()
	l.Path = location
	h := &Held{leaser: l, replicaClient: &generationFake{}}
	return h.BindGeneration(prefix, func(context.Context) error { return nil })
}

func TestGenerationClientRetentionRefusedButUploadsContinue(t *testing.T) {
	f := &generationFake{}
	c := &generationClient{client: f, check: func(context.Context) error { return nil }}
	for _, level := range []int{0, 1, litestream.SnapshotLevel} {
		if err := c.DeleteLTXFiles(t.Context(), []*ltx.FileInfo{{Level: level, MinTXID: 1, MaxTXID: 2}}); !errors.Is(err, ErrGenerationRetention) {
			t.Fatalf("level %d delete: %v", level, err)
		}
	}
	if f.calls != 0 {
		t.Fatal("bare-key deletion reached backing client")
	}
	if _, err := c.WriteLTXFile(t.Context(), 0, 3, 3, bytes.NewReader(nil)); err != nil {
		t.Fatal(err)
	}
	if f.calls != 1 {
		t.Fatal("replication stopped after refusal")
	}
}

func TestGenerationClientDisablesRemoteRetentionInRealStore(t *testing.T) {
	for _, bound := range []bool{false, true} {
		l := lss3.NewLeaser()
		l.Path = "instances/x.example/inc"
		h := &Held{leaser: l, replicaClient: &generationFake{}}
		if bound {
			if err := h.BindGeneration(l.Path+"/g-one/", func(context.Context) error { return nil }); err != nil {
				t.Fatal(err)
			}
		}
		_, db, err := h.newStore(filepath.Join(t.TempDir(), "db"), time.Second)
		if err != nil {
			t.Fatal(err)
		}
		if db.RetentionEnabled == bound {
			t.Fatalf("bound=%v retention=%v", bound, db.RetentionEnabled)
		}
	}
}

func TestGenerationClientRealRetentionMonitorsDoNotDelete(t *testing.T) {
	l := lss3.NewLeaser()
	l.Path = "instances/x.example/inc"
	f := &generationFake{files: []*ltx.FileInfo{{Level: 1, MinTXID: 1, MaxTXID: 2, CreatedAt: time.Unix(1, 0)}, {Level: 1, MinTXID: 3, MaxTXID: 4, CreatedAt: time.Unix(2, 0)}}}
	h := &Held{leaser: l, replicaClient: f}
	if err := h.BindGeneration(l.Path+"/g-one/", func(context.Context) error { return nil }); err != nil {
		t.Fatal(err)
	}
	store, db, err := h.newStore(filepath.Join(t.TempDir(), "db"), time.Second)
	if err != nil {
		t.Fatal(err)
	}
	if store.RetentionEnabled || db.RetentionEnabled {
		t.Fatal("remote retention enabled")
	}
	db.MonitorInterval = 0
	db.Replica.MonitorEnabled = false
	if err := db.Open(); err != nil {
		t.Fatal(err)
	}
	defer db.Close(t.Context())
	// These execute the real DB and compactor retention code. A deletion
	// attempt would get ErrGenerationRetention and fail the call.
	if err := db.EnforceRetentionByTXID(t.Context(), 1, 10); err != nil {
		t.Fatal(err)
	}
	if _, err := db.EnforceSnapshotRetention(t.Context(), time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := db.EnforceL0RetentionByTime(t.Context()); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Replica.Client.WriteLTXFile(t.Context(), 0, 5, 5, bytes.NewReader(nil)); err != nil {
		t.Fatal(err)
	}
}

// BindRoot gives the root layout the same per-request authority a bound
// generation has (moves Task 5 fix round 1, ruling 2): every replica request
// below the lease's own path first passes check, so an authority withdrawn
// between two requests refuses the second before it reaches storage. It is
// as immutable as BindGeneration, and the two exclude each other.
func TestBindRootChecksEveryRequest(t *testing.T) {
	leaser := lss3.NewLeaser()
	leaser.Path = "instances/shop.example"
	f := &generationFake{}
	h := &Held{leaser: leaser, replicaClient: f}
	withdrawn := errors.New("the claim moved")
	calls := 0
	check := func(context.Context) error {
		calls++
		if calls > 1 {
			return withdrawn
		}
		return nil
	}
	if err := h.BindRoot(check); err != nil {
		t.Fatal(err)
	}
	if err := h.replicaClient.Init(t.Context()); err != nil {
		t.Fatalf("first request: %v", err)
	}
	if _, err := h.replicaClient.LTXFiles(t.Context(), 0, 0, false); !errors.Is(err, withdrawn) {
		t.Fatalf("second request after the authority was withdrawn: %v", err)
	}
	if f.calls != 1 {
		t.Fatalf("storage saw %d requests, want the first only", f.calls)
	}
	if err := h.BindRoot(check); err != ErrGenerationBinding {
		t.Fatalf("rebind=%v", err)
	}
	if err := h.BindGeneration("instances/shop.example/g-a", check); err != ErrGenerationBinding {
		t.Fatalf("a generation over the root binding=%v", err)
	}
	h2 := &Held{leaser: leaser, replicaClient: &generationFake{}}
	if err := h2.BindGeneration("instances/shop.example/g-a", check); err != nil {
		t.Fatal(err)
	}
	if err := h2.BindRoot(check); err != ErrGenerationBinding {
		t.Fatalf("the root over a generation binding=%v", err)
	}
	if err := (&Held{leaser: leaser}).BindRoot(nil); err != ErrGenerationBinding {
		t.Fatalf("nil check=%v", err)
	}
}