package lease

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/benbjohnson/litestream"
)

// casS3 is one key's worth of S3 with conditional writes: If-None-Match: *
// refuses an existing object, If-Match refuses another etag, with 412.
type casS3 struct {
	mu   sync.Mutex
	reqs []string
	body []byte
	etag string
	n    int
	puts []http.Header
}

func newCasS3(t *testing.T) (*casS3, string) {
	t.Helper()
	f := &casS3{}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		f.reqs = append(f.reqs, r.Method)
		fail := func() {
			w.Header().Set("Content-Type", "application/xml")
			w.WriteHeader(http.StatusPreconditionFailed)
			_, _ = w.Write([]byte(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>PreconditionFailed</Code><Message>precondition</Message></Error>`))
		}
		switch r.Method {
		case http.MethodGet:
			if f.body == nil {
				w.Header().Set("Content-Type", "application/xml")
				w.WriteHeader(http.StatusNotFound)
				_, _ = w.Write([]byte(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>NoSuchKey</Code><Message>no such key</Message></Error>`))
				return
			}
			w.Header().Set("ETag", f.etag)
			_, _ = w.Write(f.body)
		case http.MethodPut:
			f.puts = append(f.puts, r.Header.Clone())
			if r.Header.Get("If-None-Match") == "*" && f.body != nil {
				fail()
				return
			}
			if m := r.Header.Get("If-Match"); m != "" && strings.Trim(m, `"`) != strings.Trim(f.etag, `"`) {
				fail()
				return
			}
			b, _ := io.ReadAll(r.Body)
			f.n++
			f.body, f.etag = b, fmt.Sprintf(`"etag-%d"`, f.n)
			w.Header().Set("ETag", f.etag)
		case http.MethodDelete:
			if m := r.Header.Get("If-Match"); m != "" && strings.Trim(m, `"`) != strings.Trim(f.etag, `"`) {
				fail()
				return
			}
			f.body, f.etag = nil, ""
			w.WriteHeader(http.StatusNoContent)
		}
	}))
	t.Cleanup(srv.Close)
	return f, srv.URL
}

func (f *casS3) set(t *testing.T, l litestream.Lease) {
	t.Helper()
	raw, err := json.Marshal(l)
	if err != nil {
		t.Fatal(err)
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	f.n++
	f.body, f.etag = raw, fmt.Sprintf(`"etag-%d"`, f.n)
}

func (f *casS3) methods() []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]string(nil), f.reqs...)
}

func (f *casS3) lease(t *testing.T) litestream.Lease {
	t.Helper()
	f.mu.Lock()
	defer f.mu.Unlock()
	var l litestream.Lease
	if err := json.Unmarshal(f.body, &l); err != nil {
		t.Fatalf("lock %q: %v", f.body, err)
	}
	return l
}

func casEnv(t *testing.T) (*casS3, context.Context) {
	t.Helper()
	f, url := newCasS3(t)
	fakeCreds(t)
	t.Setenv("CARLOS_S3_ENDPOINT", url)
	return f, context.Background()
}

// A released lease deletes its lock, so the next wake finds none: it
// writes the lease create-only in one round trip, the same lease
// litestream writes for an absent lock (generation 1, this owner).
func TestAcquireAnAbsentLockIsOneCreateOnlyWrite(t *testing.T) {
	f, ctx := casEnv(t)
	held, err := Acquire(ctx, "b", "inst/x.example.com", "eu-west-1")
	if err != nil {
		t.Fatal(err)
	}
	if got := f.methods(); len(got) != 1 || got[0] != http.MethodPut {
		t.Fatalf("requests = %v, want one PUT", got)
	}
	if h := f.puts[0]; h.Get("If-None-Match") != "*" || h.Get("If-Match") != "" {
		t.Fatalf("the write was not create-only: %v", h)
	}
	l := f.lease(t)
	if l.Generation != 1 || l.Owner == "" || time.Until(l.ExpiresAt) < LeaseTTL-time.Minute {
		t.Fatalf("lease = %+v, want generation 1, an owner and a full TTL", l)
	}
	// The handle carries the etag its write returned: release is
	// conditional on it, so it deletes this lock and no other.
	if err := held.Release(ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.methods(); got[len(got)-1] != http.MethodDelete {
		t.Fatalf("requests = %v, want the release's DELETE", got)
	}
}

// An expired lock is taken over as litestream takes it over: read, then a
// conditional write of the next generation.
func TestAcquireAnExpiredLockFallsBackToTheTakeover(t *testing.T) {
	f, ctx := casEnv(t)
	f.set(t, litestream.Lease{Generation: 4, ExpiresAt: time.Now().Add(-time.Minute), Owner: "gone:1"})
	held, err := Acquire(ctx, "b", "inst/x.example.com", "eu-west-1")
	if err != nil {
		t.Fatal(err)
	}
	defer held.Release(ctx)
	if l := f.lease(t); l.Generation != 5 || l.Owner == "gone:1" {
		t.Fatalf("lease = %+v, want generation 5 under this owner", l)
	}
}

// A live lock is someone else's: refused with its holder, as before, and
// left exactly as it was.
func TestAcquireALiveLockIsRefused(t *testing.T) {
	f, ctx := casEnv(t)
	f.set(t, litestream.Lease{Generation: 2, ExpiresAt: time.Now().Add(10 * time.Minute), Owner: "other:7"})
	before := f.lease(t)
	_, err := Acquire(ctx, "b", "inst/x.example.com", "eu-west-1")
	var held *HeldError
	if !errors.As(err, &held) || held.Owner != "other:7" {
		t.Fatalf("Acquire = %v, want a HeldError naming other:7", err)
	}
	if after := f.lease(t); after != before {
		t.Fatalf("lock changed: %+v -> %+v", before, after)
	}
}