package leasetest

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"testing"

	"github.com/carlosframework/platform/internal/lease"
)

// Maker builds a store under test, the model backing it, the bucket name,
// and (for wire stores) a request counter.
type Maker func(t *testing.T) (st lease.VersionedStore, m *VersionedBucket, bucket string, served func() int)

// RunVersionedStoreContract is the behaviour every lease.VersionedStore
// must have (spec §13, as covered by Plan 2).
func RunVersionedStoreContract(t *testing.T, mk Maker) {
	ctx := context.Background()

	t.Run("listing: key order, latest first, markers included, prefix bounded", func(t *testing.T) {
		st, m, b, _ := mk(t)
		m.Put(b, "p/a", []byte("one"))
		m.Put(b, "p/a", []byte("two"))
		m.BareDelete(b, "p/a")
		m.Put(b, "p/b", []byte("x"))
		m.Put(b, "q/c", []byte("other prefix"))
		vs, err := st.ListVersions(ctx, b, "p/")
		if err != nil {
			t.Fatal(err)
		}
		if len(vs) != 4 || vs[0].Key != "p/a" || !vs[0].IsLatest || !vs[0].DeleteMarker || vs[3].Key != "p/b" || !vs[3].IsLatest {
			t.Fatalf("listing = %+v", vs)
		}
		for _, v := range vs[1:3] {
			if v.Key != "p/a" || v.IsLatest || v.DeleteMarker || v.VersionID == "" {
				t.Fatalf("non-latest entries = %+v", vs[1:3])
			}
		}
	})

	t.Run("listing spans pages, including many versions of one key", func(t *testing.T) {
		st, m, b, _ := mk(t)
		m.PageSize = 3
		for i := 0; i < 5; i++ {
			m.Put(b, "p/same", []byte{byte(i)})
		}
		for i := 0; i < 4; i++ {
			m.Put(b, fmt.Sprintf("p/k%02d", i), []byte{byte(i)})
		}
		vs, err := st.ListVersions(ctx, b, "p/")
		if err != nil || len(vs) != 9 {
			t.Fatalf("paged listing = %d, %v; want 9", len(vs), err)
		}
		seen := map[string]bool{}
		for _, v := range vs {
			if seen[v.Key+"@"+v.VersionID] {
				t.Fatalf("duplicate %s@%s", v.Key, v.VersionID)
			}
			seen[v.Key+"@"+v.VersionID] = true
		}
	})

	t.Run("listing fails whole on a page error", func(t *testing.T) {
		st, m, b, _ := mk(t)
		m.PageSize = 2
		for i := 0; i < 6; i++ {
			m.Put(b, fmt.Sprintf("p/k%d", i), []byte{byte(i)})
		}
		m.FailListAfterPages = 2
		if vs, err := st.ListVersions(ctx, b, "p/"); err == nil || vs != nil {
			t.Fatalf("got %d entries and error %v; want no entries and an error", len(vs), err)
		}
	})

	t.Run("read and stream by exact version; markers and missing", func(t *testing.T) {
		st, m, b, _ := mk(t)
		k := "dir/a b+c%?d"
		v1 := m.Put(b, k, []byte("old"))
		m.Put(b, k, []byte("new"))
		if body, err := st.GetVersion(ctx, b, k, v1); err != nil || string(body) != "old" {
			t.Fatalf("GetVersion = %q, %v", body, err)
		}
		rc, err := st.OpenVersion(ctx, b, k, v1)
		if err != nil {
			t.Fatal(err)
		}
		body, _ := io.ReadAll(rc)
		rc.Close()
		if string(body) != "old" {
			t.Fatalf("OpenVersion = %q", body)
		}
		mk1 := m.BareDelete(b, k)
		if _, err := st.GetVersion(ctx, b, k, mk1); !errors.Is(err, lease.ErrIsDeleteMarker) {
			t.Fatalf("GetVersion(marker) = %v, want ErrIsDeleteMarker", err)
		}
		if _, err := st.GetVersion(ctx, b, k, "v999999"); !errors.Is(err, lease.ErrMirrorNotFound) {
			t.Fatalf("GetVersion(missing) = %v, want ErrMirrorNotFound", err)
		}
		// A version hidden behind a marker is still readable by id.
		if body, err := st.GetVersion(ctx, b, k, v1); err != nil || string(body) != "old" {
			t.Fatalf("GetVersion behind a marker = %q, %v", body, err)
		}
	})

	t.Run("range by exact version", func(t *testing.T) {
		st, m, b, served := mk(t)
		ranged, ok := st.(lease.VersionRangeOpener)
		if !ok {
			t.Skip("store does not support ranged version reads")
		}
		key := "dir/a b+c%?d"
		id := m.Put(b, key, []byte("original"))
		m.Put(b, key, []byte("replacement"))
		marker := m.BareDelete(b, key)
		for _, tc := range []struct {
			name           string
			offset, length int64
			want           string
		}{
			{"bounded", 2, 3, "igi"},
			{"to end", 3, -1, "ginal"},
			{"past end", 6, 100, "al"},
			{"whole", 0, -1, "original"},
		} {
			t.Run(tc.name, func(t *testing.T) {
				rc, err := ranged.OpenVersionRange(ctx, b, key, id, tc.offset, tc.length)
				if err != nil {
					t.Fatal(err)
				}
				body, err := io.ReadAll(rc)
				closeErr := rc.Close()
				if err != nil || closeErr != nil || string(body) != tc.want {
					t.Fatalf("range = %q, %v, close %v; want %q", body, err, closeErr, tc.want)
				}
			})
		}
		for _, tc := range []struct {
			name, bucket, version string
			offset, length        int64
			want                  error
			local                 bool
		}{
			{"missing version ID", b, "", 0, 1, lease.ErrNoVersionID, true},
			{"negative offset", b, id, -1, 1, nil, true},
			{"zero length", b, id, 0, 0, nil, true},
			{"overflow", b, id, 2, 1<<63 - 1, nil, true},
			{"unsatisfiable", b, id, 8, -1, nil, false},
			{"missing version", b, "missing", 0, 1, lease.ErrMirrorNotFound, false},
			{"marker", b, marker, 0, 1, lease.ErrIsDeleteMarker, false},
			{"missing bucket", "nobucket", id, 0, 1, nil, false},
		} {
			t.Run(tc.name, func(t *testing.T) {
				before := 0
				if served != nil {
					before = served()
				}
				rc, err := ranged.OpenVersionRange(ctx, tc.bucket, key, tc.version, tc.offset, tc.length)
				if rc != nil {
					rc.Close()
				}
				if err == nil || (tc.want != nil && !errors.Is(err, tc.want)) {
					t.Fatalf("range error = %v; want %v", err, tc.want)
				}
				if tc.want == nil && errors.Is(err, lease.ErrMirrorNotFound) {
					t.Fatalf("range error was treated as missing version: %v", err)
				}
				if tc.local && served != nil && served() != before {
					t.Fatal("invalid range sent a request")
				}
			})
		}
		if err := st.DeleteVersion(ctx, b, key, id); err != nil {
			t.Fatal(err)
		}
		if rc, err := ranged.OpenVersionRange(ctx, b, key, id, 2, 3); !errors.Is(err, lease.ErrMirrorNotFound) {
			if rc != nil {
				rc.Close()
			}
			t.Fatalf("deleted version range = %v", err)
		}
		m.FailOn = func(op, key string) error { return errors.New("offline") }
		if rc, err := ranged.OpenVersionRange(ctx, b, key, id, 2, 3); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
			if rc != nil {
				rc.Close()
			}
			t.Fatalf("failed range = %v", err)
		}
	})

	t.Run("copy by exact version; create-only; markers", func(t *testing.T) {
		st, m, b, _ := mk(t)
		k := "src/a b+c%?d"
		v1 := m.Put(b, k, []byte("old"))
		m.Put(b, k, []byte("new"))
		dv, etag, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true)
		if err != nil || dv == "" || etag == "" || strings.Contains(etag, "-") {
			t.Fatalf("CopyVersion = %q %q %v", dv, etag, err)
		}
		if cur, ok := m.Current(b, "dst/x"); !ok || string(cur) != "old" {
			t.Fatalf("copied %q", cur)
		}
		if _, _, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true); !errors.Is(err, lease.ErrDestinationExists) {
			t.Fatalf("create-only copy onto existing = %v", err)
		}
		m.BareDelete(b, "dst/x")
		if _, _, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true); err != nil {
			t.Fatalf("create-only copy over a delete marker = %v", err)
		}
		mid := m.BareDelete(b, k)
		if _, _, err := st.CopyVersion(ctx, b, k, mid, b, "dst/y", false); !errors.Is(err, lease.ErrIsDeleteMarker) {
			t.Fatalf("copy of a marker = %v, want ErrIsDeleteMarker", err)
		}
		mpID := m.PutMultipart(b, "src/mp", []byte("payload"))
		if _, etag, err := st.CopyVersion(ctx, b, "src/mp", mpID, b, "dst/mp", false); err != nil || strings.Contains(etag, "-") {
			t.Fatalf("multipart copy etag = %q, %v; want single-part", etag, err)
		}
	})

	t.Run("delete by version: exact, idempotent, never bare", func(t *testing.T) {
		st, m, b, served := mk(t)
		v1 := m.Put(b, "k", []byte("one"))
		v2 := m.Put(b, "k", []byte("two"))
		if err := st.DeleteVersion(ctx, b, "k", v2); err != nil {
			t.Fatal(err)
		}
		if cur, ok := m.Current(b, "k"); !ok || string(cur) != "one" {
			t.Fatalf("after deleting the current version: %q %v; want the older one current", cur, ok)
		}
		if err := st.DeleteVersion(ctx, b, "k", v2); err != nil {
			t.Fatalf("re-delete = %v", err)
		}
		if err := st.DeleteVersion(ctx, b, "never", "v000001"); err != nil {
			t.Fatalf("delete on a missing key = %v", err)
		}
		before, reqs := m.Snapshot(), 0
		if served != nil {
			reqs = served()
		}
		if err := st.DeleteVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
			t.Fatalf("empty version id = %v, want ErrNoVersionID", err)
		}
		if _, err := st.GetVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
			t.Fatalf("GetVersion(\"\") = %v, want ErrNoVersionID", err)
		}
		if _, err := st.OpenVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
			t.Fatalf("OpenVersion(\"\") = %v, want ErrNoVersionID", err)
		}
		if _, _, err := st.CopyVersion(ctx, b, "k", "", b, "k2", false); !errors.Is(err, lease.ErrNoVersionID) {
			t.Fatalf("CopyVersion(\"\") = %v, want ErrNoVersionID", err)
		}
		if m.Snapshot() != before {
			t.Fatal("state changed on an empty-version delete")
		}
		if served != nil && served() != reqs {
			t.Fatal("a request was sent for an empty version id")
		}
		_ = v1
	})

	t.Run("delete a noncurrent version keeps the current one", func(t *testing.T) {
		st, m, b, _ := mk(t)
		v1 := m.Put(b, "k", []byte("one"))
		v2 := m.Put(b, "k", []byte("two"))
		if err := st.DeleteVersion(ctx, b, "k", v1); err != nil {
			t.Fatal(err)
		}
		if cur, ok := m.Current(b, "k"); !ok || string(cur) != "two" || m.LatestID(b, "k") != v2 {
			t.Fatalf("current after a noncurrent delete = %q %v %s", cur, ok, m.LatestID(b, "k"))
		}
		if _, err := st.GetVersion(ctx, b, "k", v1); !errors.Is(err, lease.ErrMirrorNotFound) {
			t.Fatalf("deleted version still readable: %v", err)
		}
	})

	t.Run("a missing bucket is an error, never absence", func(t *testing.T) {
		st, _, _, _ := mk(t)
		if err := st.DeleteVersion(ctx, "nobucket", "k", "v000001"); err == nil {
			t.Error("delete in a missing bucket succeeded")
		}
		if _, err := st.GetVersion(ctx, "nobucket", "k", "v000001"); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
			t.Errorf("get in a missing bucket = %v", err)
		}
		if _, err := st.ListVersions(ctx, "nobucket", ""); err == nil {
			t.Error("list of a missing bucket succeeded")
		}
	})

	t.Run("put if absent", func(t *testing.T) {
		st, m, b, _ := mk(t)
		if v, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); err != nil || v == "" {
			t.Fatalf("first = %q %v", v, err)
		}
		if _, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); !errors.Is(err, lease.ErrExists) {
			t.Fatalf("second = %v, want ErrExists", err)
		}
		m.BareDelete(b, "m/closed.json")
		if _, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); err != nil {
			t.Fatalf("over a marker = %v", err)
		}
	})

	t.Run("injected failures surface as errors from every operation", func(t *testing.T) {
		st, m, b, _ := mk(t)
		v := m.Put(b, "k", []byte("x"))
		boom := errors.New("boom")
		m.FailOn = func(op, key string) error { return boom }
		if _, err := st.ListVersions(ctx, b, ""); err == nil {
			t.Error("list: no error")
		}
		if _, err := st.GetVersion(ctx, b, "k", v); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
			t.Errorf("get: %v", err)
		}
		if _, _, err := st.CopyVersion(ctx, b, "k", v, b, "k2", false); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
			t.Errorf("copy: %v", err)
		}
		if err := st.DeleteVersion(ctx, b, "k", v); err == nil {
			t.Error("delete: no error")
		}
		if _, err := st.PutIfAbsent(ctx, b, "new", []byte("x")); err == nil {
			t.Error("put: no error")
		}
		m.FailOn = nil
		if _, ok := m.Current(b, "k"); !ok {
			t.Fatal("a failed delete removed data")
		}
	})
}