package release

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"os"
	"testing"
	"time"

	"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"
)

// TestS3StoreMinioContract runs release.S3Store against a real S3-compatible
// server: the conditional GET (a 304 for the current ETag), the ETag listing
// agreeing with Get, and the change index bump through every S3Store
// mutation path. A fake client cannot show any of these: the 304 and the
// ETag forms are the server's. Skipped unless CARLOS_MINIO_TEST=1;
// .amadan/ci.d/30-minio-contract stands the server up and requires this
// test's PASS line, so a skip cannot pass for coverage there.
func TestS3StoreMinioContract(t *testing.T) {
	if os.Getenv("CARLOS_MINIO_TEST") != "1" {
		t.Skip("set CARLOS_MINIO_TEST=1 (with a running Minio and the MINIO_* env vars) to run the S3Store contract test; see .amadan/ci.d/30-minio-contract, which stands one up")
	}
	endpoint := minioEnv(t, "MINIO_ENDPOINT")
	key := minioEnv(t, "MINIO_SCOPED_KEY")
	secret := minioEnv(t, "MINIO_SCOPED_SECRET")
	bucket := minioEnv(t, "MINIO_BUCKET")

	client := awss3.New(awss3.Options{
		Region:                     "us-east-1",
		Credentials:                credentials.NewStaticCredentialsProvider(key, secret, ""),
		BaseEndpoint:               aws.String(endpoint),
		UsePathStyle:               true,
		RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
		ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
	})
	s := &S3Store{Client: client, Bucket: bucket}
	s.EnableChangeIndex()
	ctx := context.Background()

	// One scope per run, so nothing else in the shared bucket moves its
	// generation. control/changes.json is the one shared key: the test
	// compares generations before and after and leaves it in place.
	app := fmt.Sprintf("mt%d", time.Now().UnixNano())
	scope := "apps/" + app + "/shop"
	base := scope + "/"
	var created []string
	t.Cleanup(func() {
		for _, k := range created {
			_ = s.deleteRaw(context.Background(), k) // not Delete: no bump on the way out
		}
	})
	keyA, keyB := base+"a.json", base+"b.json"
	created = append(created, keyA, keyB)

	gen := func() uint64 {
		t.Helper()
		idx, _, err := ReadChangeIndex(ctx, s) // an absent index reads as empty
		if err != nil {
			t.Fatalf("read change index: %v", err)
		}
		return idx.Entries[scope]
	}
	bumped := func(name string, before uint64) uint64 {
		t.Helper()
		after := gen()
		if after != before+1 {
			t.Fatalf("%s: generation %d -> %d, want exactly +1", name, before, after)
		}
		return after
	}

	t.Run("conditional get and etag listing", func(t *testing.T) {
		etag, err := s.putRaw(ctx, keyA, []byte("one"), "")
		if err != nil {
			t.Fatal(err)
		}
		body, e, changed, err := s.GetIfChanged(ctx, keyA, "")
		if err != nil || !changed || string(body) != "one" || e != etag {
			t.Fatalf("empty etag: %q %q (put %q) %v %v", body, e, etag, changed, err)
		}
		body, e2, changed, err := s.GetIfChanged(ctx, keyA, e)
		if err != nil || changed || body != nil || e2 != e {
			t.Fatalf("current etag must be a 304: %q %q %v %v", body, e2, changed, err)
		}
		if _, err := s.putRaw(ctx, keyA, []byte("two"), e); err != nil {
			t.Fatal(err)
		}
		body, e3, changed, err := s.GetIfChanged(ctx, keyA, e)
		if err != nil || !changed || string(body) != "two" || e3 == e {
			t.Fatalf("after rewrite: %q %q %v %v", body, e3, changed, err)
		}
		if _, _, _, err := s.GetIfChanged(ctx, base+"absent.json", e); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("absent: %v", err)
		}
		if _, err := s.putRaw(ctx, keyB, []byte("b"), ""); err != nil {
			t.Fatal(err)
		}
		listed, err := s.ListETags(ctx, base)
		if err != nil {
			t.Fatal(err)
		}
		if len(listed) != 2 || listed[0].Key != keyA || listed[1].Key != keyB {
			t.Fatalf("listed %v, want %s and %s sorted", listed, keyA, keyB)
		}
		for _, ke := range listed {
			_, ge, err := s.Get(ctx, ke.Key)
			if err != nil || ge != ke.ETag {
				t.Fatalf("%s: listed etag %q, Get etag %q (%v)", ke.Key, ke.ETag, ge, err)
			}
		}
		for _, k := range []string{keyA, keyB} {
			if err := s.deleteRaw(ctx, k); err != nil {
				t.Fatal(err)
			}
		}
	})

	t.Run("bump through every mutation path", func(t *testing.T) {
		g := gen()
		etag, err := s.Put(ctx, keyA, []byte("one"), "")
		if err != nil {
			t.Fatal(err)
		}
		g = bumped("Put", g)

		// A Put whose ifMatch is stale loses and must not bump.
		if _, err := s.Put(ctx, keyA, []byte("stale"), `"00000000000000000000000000000000"`); !errors.Is(err, ErrCASConflict) {
			t.Fatalf("stale Put: %v", err)
		}
		if after := gen(); after != g {
			t.Fatalf("a refused Put moved the generation %d -> %d", g, after)
		}

		etag, err = s.PutFrom(ctx, keyA, bytes.NewReader([]byte("two")), 3, etag)
		if err != nil {
			t.Fatal(err)
		}
		g = bumped("PutFrom", g)

		if err := DeleteIfMatch(ctx, s, keyA, etag); err != nil {
			t.Fatal(err)
		}
		g = bumped("DeleteIfMatch", g)

		if _, err := s.Put(ctx, keyA, []byte("three"), ""); err != nil {
			t.Fatal(err)
		}
		g = bumped("Put again", g)
		if err := s.Delete(ctx, keyA); err != nil {
			t.Fatal(err)
		}
		bumped("Delete", g)
	})
}

func minioEnv(t *testing.T, name string) string {
	t.Helper()
	v := os.Getenv(name)
	if v == "" {
		t.Fatalf("%s is not set (required when CARLOS_MINIO_TEST=1)", name)
	}
	return v
}