package leasetest

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"testing"

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

// failingReader yields some bytes and then an error, like a dropped upload.
type failingReader struct{ r io.Reader }

func (f *failingReader) Read(p []byte) (int, error) {
	n, err := f.r.Read(p)
	if err == io.EOF {
		return n, errors.New("connection reset")
	}
	return n, err
}

func TestS3ServerFailedUploadStoresNothing(t *testing.T) {
	m := NewVersionedBucket()
	m.AddBucket("data")
	h, _ := NewS3Server(m)
	before := m.Snapshot()
	req := httptest.NewRequest(http.MethodPut, "/data/k", &failingReader{strings.NewReader("partial")})
	req.Header.Set("If-None-Match", "*")
	rec := httptest.NewRecorder()
	h.ServeHTTP(rec, req)
	if rec.Code < 400 {
		t.Fatalf("failed upload answered %d", rec.Code)
	}
	if m.Snapshot() != before {
		t.Fatalf("failed upload changed the bucket:\n%s", m.Snapshot())
	}
}

func do(t *testing.T, h http.Handler, method, path string, body string, hdr map[string]string) *httptest.ResponseRecorder {
	t.Helper()
	req := httptest.NewRequest(method, path, strings.NewReader(body))
	for k, v := range hdr {
		req.Header.Set(k, v)
	}
	rec := httptest.NewRecorder()
	h.ServeHTTP(rec, req)
	return rec
}

// A plain PUT (no If-None-Match) stores a new version beside the existing
// one, as S3 with versioning on does; the create-only PUT keeps refusing
// an existing current object. A plain DELETE adds a delete marker.
func TestS3ServerPlainPutCreatesAVersion(t *testing.T) {
	m := NewVersionedBucket()
	m.AddBucket("data")
	h, _ := NewS3Server(m)
	first := do(t, h, http.MethodPut, "/data/k", "one", nil)
	second := do(t, h, http.MethodPut, "/data/k", "two", nil)
	if first.Code != 200 || second.Code != 200 {
		t.Fatalf("plain puts: %d %d", first.Code, second.Code)
	}
	v1, v2 := first.Header().Get("x-amz-version-id"), second.Header().Get("x-amz-version-id")
	if v1 == "" || v2 == "" || v1 == v2 {
		t.Fatalf("version ids %q %q", v1, v2)
	}
	if got, _ := m.Current("data", "k"); string(got) != "two" {
		t.Fatalf("current %q", got)
	}
	if old, err := m.GetVersion(context.Background(), "data", "k", v1); err != nil || string(old) != "one" {
		t.Fatalf("the first version is gone: %q %v", old, err)
	}
	// a GET naming no version reads the current one
	if rec := do(t, h, http.MethodGet, "/data/k", "", nil); rec.Code != 200 || rec.Body.String() != "two" {
		t.Fatalf("plain get: %d %q", rec.Code, rec.Body.String())
	}
	if rec := do(t, h, http.MethodGet, "/data/absent", "", nil); rec.Code != http.StatusNotFound {
		t.Fatalf("plain get of an absent key: %d", rec.Code)
	}
	// the create-only path is unchanged
	if rec := do(t, h, http.MethodPut, "/data/k", "three", map[string]string{"If-None-Match": "*"}); rec.Code != http.StatusPreconditionFailed {
		t.Fatalf("create-only over a current object: %d", rec.Code)
	}
	if rec := do(t, h, http.MethodPut, "/data/fresh", "x", map[string]string{"If-None-Match": "*"}); rec.Code != 200 || rec.Header().Get("x-amz-version-id") == "" {
		t.Fatalf("create-only on a fresh key: %d", rec.Code)
	}
	// a bare DELETE leaves a delete marker, the versions stay
	if rec := do(t, h, http.MethodDelete, "/data/k", "", nil); rec.Code != http.StatusNoContent {
		t.Fatalf("bare delete: %d", rec.Code)
	}
	if _, ok := m.Current("data", "k"); ok {
		t.Fatal("the key still reads after a bare delete")
	}
	if old, err := m.GetVersion(context.Background(), "data", "k", v2); err != nil || string(old) != "two" {
		t.Fatalf("a bare delete removed a version: %q %v", old, err)
	}
}

// A copy that names no source version copies the source's current one, and
// honors x-amz-copy-source-if-match against its ETag (S3's behavior; the
// staging mirror copies this way).
func TestS3ServerCopyOfCurrentVersion(t *testing.T) {
	m := NewVersionedBucket()
	m.AddBucket("src")
	m.AddBucket("dst")
	m.Put("src", "a/b c+d", []byte("body"))
	h, _ := NewS3Server(m)
	src := "/src/a/b%20c%2Bd"
	rec := do(t, h, http.MethodPut, "/dst/out", "", map[string]string{"x-amz-copy-source": src, "x-amz-copy-source-if-match": `"` + md5hex([]byte("body")) + `"`})
	if rec.Code != 200 || rec.Header().Get("x-amz-version-id") == "" || !strings.Contains(rec.Body.String(), md5hex([]byte("body"))) {
		t.Fatalf("copy: %d %q %v", rec.Code, rec.Body.String(), rec.Header())
	}
	rec = do(t, h, http.MethodPut, "/dst/out2", "", map[string]string{"x-amz-copy-source": src, "x-amz-copy-source-if-match": `"deadbeef"`})
	if rec.Code != http.StatusPreconditionFailed || !strings.Contains(rec.Body.String(), "<Code>PreconditionFailed</Code>") {
		t.Fatalf("stale if-match: %d %q", rec.Code, rec.Body.String())
	}
	if _, ok := m.Current("dst", "out2"); ok {
		t.Fatal("a failed precondition copied")
	}
}

// Real S3 rejects a PUT whose body is not the length the request declares,
// or whose sha256 is not the one the request signed, and stores nothing. The
// fake must do the same, or a client that sends a short or altered body looks
// like it succeeded (a transport can hand the server a partial body before it
// fails).
func TestS3ServerValidatesPutBody(t *testing.T) {
	sum := sha256.Sum256([]byte("abcdefghij"))
	good := hex.EncodeToString(sum[:])
	wrong := hex.EncodeToString(make([]byte, 32))
	put := func(h http.Handler, key, body string, declared int64, sha string, extra map[string]string) *httptest.ResponseRecorder {
		req := httptest.NewRequest(http.MethodPut, "/data/"+key, strings.NewReader(body))
		req.ContentLength = declared
		if sha != "" {
			req.Header.Set("x-amz-content-sha256", sha)
		}
		for k, v := range extra {
			req.Header.Set(k, v)
		}
		rec := httptest.NewRecorder()
		h.ServeHTTP(rec, req)
		return rec
	}
	for _, c := range []struct {
		name     string
		body     string
		declared int64
		sha      string
		code     string // empty: accepted
	}{
		{"exact", "abcdefghij", 10, good, ""},
		{"unsigned payload", "abcdefghij", 10, "UNSIGNED-PAYLOAD", ""},
		{"no sha header", "abcdefghij", 10, "", ""},
		{"unknown length", "abcdefghij", -1, good, ""},
		{"short body", "abc", 10, good, "IncompleteBody"},
		{"short body unsigned", "abc", 10, "UNSIGNED-PAYLOAD", "IncompleteBody"},
		{"long body", "abcdefghijkl", 10, "", "IncompleteBody"},
		{"wrong sha", "abcdefghij", 10, wrong, "XAmzContentSHA256Mismatch"},
		{"altered body", "abcdefghiX", 10, good, "XAmzContentSHA256Mismatch"},
	} {
		for _, createOnly := range []bool{false, true} {
			m := NewVersionedBucket()
			m.AddBucket("data")
			h, _ := NewS3Server(m)
			extra := map[string]string{}
			if createOnly {
				extra["If-None-Match"] = "*"
			}
			rec := put(h, "k", c.body, c.declared, c.sha, extra)
			_, stored := m.Current("data", "k")
			if c.code == "" {
				if rec.Code != 200 || !stored {
					t.Errorf("%s (create-only %v): %d, stored %v", c.name, createOnly, rec.Code, stored)
				}
				continue
			}
			if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "<Code>"+c.code+"</Code>") {
				t.Errorf("%s (create-only %v): %d %q, want 400 %s", c.name, createOnly, rec.Code, rec.Body.String(), c.code)
			}
			if stored {
				t.Errorf("%s (create-only %v): a rejected PUT stored the body", c.name, createOnly)
			}
		}
	}
}

// A copy carries no body: its Content-Length must be what arrives and its
// signed sha256 (when a digest) the hash of nothing. A copy that fails the
// check stores nothing.
func TestS3ServerValidatesCopyBody(t *testing.T) {
	empty := sha256.Sum256(nil)
	m := NewVersionedBucket()
	m.AddBucket("src")
	m.AddBucket("dst")
	m.Put("src", "a", []byte("body"))
	h, _ := NewS3Server(m)
	copyWith := func(dst, body string, declared int64, sha string) *httptest.ResponseRecorder {
		req := httptest.NewRequest(http.MethodPut, "/dst/"+dst, strings.NewReader(body))
		req.ContentLength = declared
		req.Header.Set("x-amz-copy-source", "/src/a")
		if sha != "" {
			req.Header.Set("x-amz-content-sha256", sha)
		}
		rec := httptest.NewRecorder()
		h.ServeHTTP(rec, req)
		return rec
	}
	if rec := copyWith("ok", "", 0, hex.EncodeToString(empty[:])); rec.Code != 200 {
		t.Fatalf("a well-formed copy: %d %q", rec.Code, rec.Body.String())
	}
	if rec := copyWith("ok2", "", 0, "UNSIGNED-PAYLOAD"); rec.Code != 200 {
		t.Fatalf("an unsigned-payload copy: %d %q", rec.Code, rec.Body.String())
	}
	if rec := copyWith("long", "xyz", 0, ""); rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "<Code>IncompleteBody</Code>") {
		t.Fatalf("a copy with a body longer than declared: %d %q", rec.Code, rec.Body.String())
	}
	if rec := copyWith("sha", "", 0, hex.EncodeToString(make([]byte, 32))); rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "<Code>XAmzContentSHA256Mismatch</Code>") {
		t.Fatalf("a copy with the wrong sha256: %d %q", rec.Code, rec.Body.String())
	}
	for _, k := range []string{"long", "sha"} {
		if _, ok := m.Current("dst", k); ok {
			t.Fatalf("a rejected copy stored %q", k)
		}
	}
}

func TestS3ServerConditionalReplaceUsesReadETag(t *testing.T) {
	m := NewVersionedBucket()
	m.AddBucket("data")
	h, _ := NewS3Server(m)
	first := do(t, h, http.MethodPut, "/data/lock.json", "expired", map[string]string{"If-None-Match": "*"})
	read := do(t, h, http.MethodGet, "/data/lock.json", "", nil)
	etag := read.Header().Get("ETag")
	if etag == "" || etag != first.Header().Get("ETag") {
		t.Fatal("GET did not return current ETag", etag)
	}
	if got := do(t, h, http.MethodPut, "/data/lock.json", "new-owner", map[string]string{"If-Match": etag}); got.Code != 200 {
		t.Fatal("takeover", got.Code)
	}
	before := m.Snapshot()
	if got := do(t, h, http.MethodPut, "/data/lock.json", "stale-owner", map[string]string{"If-Match": etag}); got.Code != http.StatusPreconditionFailed {
		t.Fatal("stale CAS accepted", got.Code)
	}
	if m.Snapshot() != before {
		t.Fatal("stale CAS mutated versions")
	}
	if got := do(t, h, http.MethodPut, "/data/missing", "new", map[string]string{"If-Match": etag}); got.Code != http.StatusPreconditionFailed {
		t.Fatal("CAS on missing accepted", got.Code)
	}
	do(t, h, http.MethodDelete, "/data/lock.json", "", nil)
	if got := do(t, h, http.MethodPut, "/data/lock.json", "new", map[string]string{"If-Match": etag}); got.Code != http.StatusPreconditionFailed {
		t.Fatal("CAS over delete marker accepted", got.Code)
	}
}

// DeleteObject honours If-Match as S3 does: a stale ETag is a 412 and removes
// nothing, a missing key is a 404, the current ETag deletes, and a plain
// delete (no header) is unchanged.
func TestS3ServerConditionalDeleteUsesReadETag(t *testing.T) {
	m := NewVersionedBucket()
	m.AddBucket("data")
	h, _ := NewS3Server(m)
	do(t, h, http.MethodPut, "/data/lock.json", "one", map[string]string{"If-None-Match": "*"})
	stale := do(t, h, http.MethodGet, "/data/lock.json", "", nil).Header().Get("ETag")
	do(t, h, http.MethodPut, "/data/lock.json", "two", map[string]string{"If-Match": stale})
	current := do(t, h, http.MethodGet, "/data/lock.json", "", nil).Header().Get("ETag")
	before := m.Snapshot()
	if got := do(t, h, http.MethodDelete, "/data/lock.json", "", map[string]string{"If-Match": stale}); got.Code != http.StatusPreconditionFailed {
		t.Fatal("stale conditional delete accepted", got.Code)
	}
	if m.Snapshot() != before {
		t.Fatal("stale conditional delete mutated versions")
	}
	if got := do(t, h, http.MethodDelete, "/data/lock.json", "", map[string]string{"If-Match": current}); got.Code != http.StatusNoContent {
		t.Fatal("current conditional delete refused", got.Code)
	}
	if got := do(t, h, http.MethodDelete, "/data/lock.json", "", map[string]string{"If-Match": current}); got.Code != http.StatusNotFound {
		t.Fatal("conditional delete of a deleted key", got.Code)
	}
	if got := do(t, h, http.MethodDelete, "/data/never", "", map[string]string{"If-Match": current}); got.Code != http.StatusNotFound {
		t.Fatal("conditional delete of a missing key", got.Code)
	}
	if got := do(t, h, http.MethodDelete, "/data/never", "", nil); got.Code != http.StatusNoContent {
		t.Fatal("plain delete changed", got.Code)
	}
}

// listV2 decodes one ListObjectsV2 page from the raw wire.
func listV2(t *testing.T, h http.Handler, query string) (listBucketResult, int) {
	t.Helper()
	rec := do(t, h, http.MethodGet, "/data?list-type=2"+query, "", nil)
	var res listBucketResult
	if rec.Code == http.StatusOK {
		if err := xml.Unmarshal(rec.Body.Bytes(), &res); err != nil {
			t.Fatalf("decode %q: %v", rec.Body.String(), err)
		}
	}
	return res, rec.Code
}

// ListObjectsV2 on the wire: the current objects under a prefix in key
// order, each with its size and quoted ETag; a key whose newest version is
// a delete marker, an older version and a sibling prefix are not listed;
// max-keys pages, and the continuation token resumes after the page;
// start-after skips; a bad token and a negative max-keys are 400s; a
// delimiter rolls keys into common prefixes, which page like keys.
func TestS3ServerListObjectsV2(t *testing.T) {
	m := NewVersionedBucket()
	m.AddBucket("data")
	h, _ := NewS3Server(m)
	for _, k := range []string{"p/c", "p/a", "p/b/x", "p/d", "q/a", "p2/a"} {
		m.Put("data", k, []byte("body of "+k))
	}
	m.Put("data", "p/a", []byte("newer"))
	m.BareDelete("data", "p/d")
	res, code := listV2(t, h, "&prefix=p/")
	if code != http.StatusOK {
		t.Fatal(code)
	}
	var keys []string
	for _, o := range res.Contents {
		keys = append(keys, o.Key)
	}
	if strings.Join(keys, ",") != "p/a,p/b/x,p/c" || res.IsTruncated || res.KeyCount != 3 || res.NextContinuationToken != "" {
		t.Fatalf("listing %+v", res)
	}
	if o := res.Contents[0]; o.Size != int64(len("newer")) || o.ETag != `"`+md5hex([]byte("newer"))+`"` {
		t.Fatalf("p/a listed as %+v", o)
	}

	first, _ := listV2(t, h, "&prefix=p/&max-keys=2")
	if len(first.Contents) != 2 || !first.IsTruncated || first.NextContinuationToken == "" || first.Contents[1].Key != "p/b/x" {
		t.Fatalf("first page %+v", first)
	}
	second, _ := listV2(t, h, "&prefix=p/&max-keys=2&continuation-token="+url.QueryEscape(first.NextContinuationToken))
	if len(second.Contents) != 1 || second.Contents[0].Key != "p/c" || second.IsTruncated || second.ContinuationToken != first.NextContinuationToken {
		t.Fatalf("second page %+v", second)
	}
	after, _ := listV2(t, h, "&prefix=p/&start-after=p/a")
	if len(after.Contents) != 2 || after.Contents[0].Key != "p/b/x" {
		t.Fatalf("start-after %+v", after)
	}
	if _, code := listV2(t, h, "&continuation-token=%21%21"); code != http.StatusBadRequest {
		t.Fatal("bad token", code)
	}
	if _, code := listV2(t, h, "&max-keys=-1"); code != http.StatusBadRequest {
		t.Fatal("negative max-keys", code)
	}
	// A delimiter rolls p/b/x into the common prefix p/b/, which counts
	// against max-keys and can end a page.
	m.Put("data", "p/b/y", []byte("y"))
	d1, _ := listV2(t, h, "&prefix=p/&delimiter=/&max-keys=2")
	if len(d1.Contents) != 1 || d1.Contents[0].Key != "p/a" || len(d1.CommonPrefixes) != 1 || d1.CommonPrefixes[0].Prefix != "p/b/" ||
		d1.KeyCount != 2 || !d1.IsTruncated || d1.Delimiter != "/" {
		t.Fatalf("delimited first page %+v", d1)
	}
	d2, _ := listV2(t, h, "&prefix=p/&delimiter=/&max-keys=2&continuation-token="+url.QueryEscape(d1.NextContinuationToken))
	if len(d2.Contents) != 1 || d2.Contents[0].Key != "p/c" || len(d2.CommonPrefixes) != 0 || d2.IsTruncated {
		t.Fatalf("delimited second page %+v", d2)
	}
	top, _ := listV2(t, h, "&delimiter=/")
	var tops []string
	for _, p := range top.CommonPrefixes {
		tops = append(tops, p.Prefix)
	}
	if len(top.Contents) != 0 || strings.Join(tops, ",") != "p/,p2/,q/" {
		t.Fatalf("top level %+v", top)
	}
}

// The SDK's own paginator, through lease.NewS3Mirror, reads every page:
// 2,500 keys at the default 1,000 a page, and again at a PageSize of 7.
func TestS3ServerListObjectsV2PaginatesTheSDK(t *testing.T) {
	m := NewVersionedBucket()
	m.AddBucket("data")
	h, _ := NewS3Server(m)
	srv := httptest.NewServer(h)
	defer srv.Close()
	t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
	t.Setenv("AWS_SESSION_TOKEN", "")
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
	t.Setenv("AWS_CONFIG_FILE", t.TempDir()+"/absent")
	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", t.TempDir()+"/absent")
	t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
	want := map[string]bool{}
	for i := range 2500 {
		k := fmt.Sprintf("big/%05d", i)
		m.Put("data", k, []byte(k))
		want[k] = true
	}
	m.Put("data", "bigger/x", []byte("x")) // a sibling prefix, never listed
	ctx := context.Background()
	mirror, err := lease.NewS3Mirror(ctx, "eu-west-1", "eu-west-1")
	if err != nil {
		t.Fatal(err)
	}
	for _, page := range []int{1000, 7} {
		m.PageSize = page
		got, err := mirror.List(ctx, "data", "big/")
		if err != nil {
			t.Fatalf("page size %d: %v", page, err)
		}
		if len(got) != len(want) {
			t.Fatalf("page size %d: listed %d keys, want %d", page, len(got), len(want))
		}
		for k, o := range got {
			if !want[k] || o.Size != int64(len(k)) || o.ETag != md5hex([]byte(k)) {
				t.Fatalf("page size %d: %s listed as %+v", page, k, o)
			}
		}
	}
}