package lease_test

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"testing"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/credentials"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"

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

func wireClient(t *testing.T, url string) *awss3.Client {
	t.Helper()
	return awss3.New(awss3.Options{
		Region: "eu-west-1", BaseEndpoint: aws.String(url), UsePathStyle: true,
		Credentials:      credentials.NewStaticCredentialsProvider("k", "s", ""),
		RetryMaxAttempts: 1,
	})
}

// wireStore is a versionedS3 whose buckets "data" and "other" are served
// by two separate servers over one model, so a wrong-region routing is a
// visible failure.
func wireStore(t *testing.T) (lease.VersionedStore, *leasetest.VersionedBucket, func() int) {
	t.Helper()
	m := leasetest.NewVersionedBucket()
	m.AddBucket("data")
	m.AddBucket("other")
	h1, n1 := leasetest.NewS3Server(m, "data", "nobucket")
	h2, n2 := leasetest.NewS3Server(m, "other")
	s1, s2 := httptest.NewServer(h1), httptest.NewServer(h2)
	t.Cleanup(s1.Close)
	t.Cleanup(s2.Close)
	// "nobucket" is bound but absent from the model: the server answers
	// NoSuchBucket, for the missing-bucket contract case.
	st := lease.NewVersionedS3WithClients(map[string]*awss3.Client{
		"data": wireClient(t, s1.URL), "other": wireClient(t, s2.URL),
		"nobucket": wireClient(t, s1.URL),
	})
	return st, m, func() int { return n1() + n2() }
}

func TestVersionedS3Contract(t *testing.T) {
	leasetest.RunVersionedStoreContract(t, func(t *testing.T) (lease.VersionedStore, *leasetest.VersionedBucket, string, func() int) {
		st, m, served := wireStore(t)
		return st, m, "data", served
	})
}

// Each bucket reaches only its own endpoint: a client bound to the wrong
// endpoint gets 301 and the operation fails.
func TestVersionedS3RoutesEachBucketToItsEndpoint(t *testing.T) {
	ctx := context.Background()
	st, m, _ := wireStore(t)
	for _, b := range []string{"data", "other"} {
		v := m.Put(b, "k", []byte(b))
		if vs, err := st.ListVersions(ctx, b, ""); err != nil || len(vs) != 1 {
			t.Fatalf("list %s = %v, %v", b, vs, err)
		}
		if err := st.DeleteVersion(ctx, b, "k", v); err != nil {
			t.Fatalf("delete %s = %v", b, err)
		}
	}
	m2 := leasetest.NewVersionedBucket()
	m2.AddBucket("data")
	h, _ := leasetest.NewS3Server(m2, "other")
	srv := httptest.NewServer(h)
	t.Cleanup(srv.Close)
	wrong := lease.NewVersionedS3WithClients(map[string]*awss3.Client{"data": wireClient(t, srv.URL)})
	if _, err := wrong.ListVersions(ctx, "data", ""); err == nil {
		t.Fatal("a wrong-region endpoint answered")
	}
}

func TestVersionedS3RefusesUnboundBucket(t *testing.T) {
	st, _, _ := wireStore(t)
	if _, err := st.ListVersions(context.Background(), "unbound", ""); err == nil {
		t.Fatal("listing an unbound bucket did not fail")
	}
}

func TestVersionedS3CopyEscapesLikeAWS(t *testing.T) {
	st, m, _ := wireStore(t)
	key := "src/a b+c%?d"
	v1 := m.Put("data", key, []byte("old"))
	m.Put("data", key, []byte("new"))
	if _, _, err := st.CopyVersion(context.Background(), "data", key, v1, "other", "dst/k", true); err != nil {
		t.Fatal(err)
	}
	if cur, ok := m.Current("other", "dst/k"); !ok || string(cur) != "old" {
		t.Fatalf("copied %q, want the pinned old version", cur)
	}
}

func TestVersionedS3RangeWire(t *testing.T) {
	m := leasetest.NewVersionedBucket()
	m.AddBucket("data")
	key := "dir/a b+c%?d"
	id := m.Put("data", key, []byte("original"))
	m.Put("data", key, []byte("replacement"))
	handler, _ := leasetest.NewS3Server(m, "data")
	for _, tc := range []struct {
		length                     int64
		header, contentRange, body string
	}{
		{3, "bytes=2-4", "bytes 2-4/8", "igi"},
		{-1, "bytes=2-", "bytes 2-7/8", "iginal"},
	} {
		t.Run(tc.header, func(t *testing.T) {
			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if r.Method != http.MethodGet || r.URL.Query().Get("versionId") != id || r.Header.Get("Range") != tc.header {
					t.Errorf("request = %s %s, Range %q", r.Method, r.URL, r.Header.Get("Range"))
				}
				rec := httptest.NewRecorder()
				handler.ServeHTTP(rec, r)
				if rec.Code != http.StatusPartialContent || rec.Header().Get("Content-Range") != tc.contentRange || rec.Body.String() != tc.body {
					t.Errorf("response = %d %v %q", rec.Code, rec.Header(), rec.Body.String())
				}
				for k, values := range rec.Header() {
					w.Header()[k] = values
				}
				w.WriteHeader(rec.Code)
				w.Write(rec.Body.Bytes())
			}))
			defer srv.Close()
			st := lease.NewVersionedS3WithClients(map[string]*awss3.Client{"data": wireClient(t, srv.URL)})
			ranged, ok := st.(lease.VersionRangeOpener)
			if !ok {
				t.Fatal("S3 store lacks ranged version reads")
			}
			rc, err := ranged.OpenVersionRange(t.Context(), "data", key, id, 2, tc.length)
			if err != nil {
				t.Fatal(err)
			}
			body, err := io.ReadAll(rc)
			rc.Close()
			if err != nil || string(body) != tc.body {
				t.Fatalf("read = %q, %v", body, err)
			}
		})
	}
}

// PutFileIfAbsent streams a file from disk create-only: the object lands
// with the file's bytes and a version id, and a second put is ErrExists
// with the first object untouched.
func TestVersionedS3PutFileIfAbsent(t *testing.T) {
	ctx := context.Background()
	st, m, _ := wireStore(t)
	fp, ok := st.(lease.FilePutIfAbsent)
	if !ok {
		t.Fatal("the S3 store cannot put a file create-only")
	}
	path := filepath.Join(t.TempDir(), "sealed")
	if err := os.WriteFile(path, []byte("sealed bytes"), 0o600); err != nil {
		t.Fatal(err)
	}
	v, err := fp.PutFileIfAbsent(ctx, "data", "move-snapshots/h/op/files-ab.sealed", path)
	if err != nil || v == "" {
		t.Fatalf("first put = %q, %v", v, err)
	}
	if cur, ok := m.Current("data", "move-snapshots/h/op/files-ab.sealed"); !ok || string(cur) != "sealed bytes" {
		t.Fatalf("stored %q", cur)
	}
	if err := os.WriteFile(path, []byte("other bytes"), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := fp.PutFileIfAbsent(ctx, "data", "move-snapshots/h/op/files-ab.sealed", path); !errors.Is(err, lease.ErrExists) {
		t.Fatalf("second put = %v, want ErrExists", err)
	}
	if cur, _ := m.Current("data", "move-snapshots/h/op/files-ab.sealed"); string(cur) != "sealed bytes" {
		t.Fatalf("the second put overwrote: %q", cur)
	}
}