package lease

// CARLOS_S3_ENDPOINT tests. Unlike lease_test.go these need no AWS and no
// CARLOS_TEST_S3_BUCKET: the point of the override is that S3 traffic can be
// pointed anywhere, so the test points it at an httptest server and asserts
// the request actually arrives. A regression here — the leaser silently going
// back to real AWS — is exactly what cost this deployment hibernation and the
// whole `carlos instances create` provisioning path.

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

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/benbjohnson/litestream"
	lss3 "github.com/benbjohnson/litestream/s3"
	"github.com/carlosframework/platform/internal/rootaws"
)

// fakeS3 is just enough of S3 for the leaser's three verbs (GET/PUT/DELETE on
// one key) to complete a full acquire-then-release, recording every request so
// a test can assert on where the traffic went.
type fakeS3 struct {
	mu   sync.Mutex
	reqs []string // "METHOD /path"
	body []byte   // last PUT body
}

func (f *fakeS3) record(method, path string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.reqs = append(f.reqs, method+" "+path)
}

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

func (f *fakeS3) putBody() []byte {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]byte(nil), f.body...)
}

func newFakeS3(t *testing.T) (*fakeS3, string) {
	t.Helper()
	f := &fakeS3{}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		f.record(r.Method, r.URL.Path)
		switch r.Method {
		case http.MethodGet:
			// No lease on the key yet. The XML body matters: litestream
			// classifies "absent" by the S3 error CODE, and a bare 404 with no
			// body would read as a hard failure instead.
			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>`))
		case http.MethodPut:
			b, _ := io.ReadAll(r.Body)
			f.mu.Lock()
			f.body = b
			f.mu.Unlock()
			w.Header().Set("ETag", `"lease-etag"`)
			w.WriteHeader(http.StatusOK)
		case http.MethodDelete:
			w.WriteHeader(http.StatusNoContent)
		default:
			w.WriteHeader(http.StatusMethodNotAllowed)
		}
	}))
	t.Cleanup(srv.Close)
	return f, srv.URL
}

// fakeCreds plants credentials the AWS default chain will resolve without
// touching the operator's real profile or IMDS — the same isolation instinct
// as carlosTestEnv, applied to the SDK instead of the carlos binary.
func fakeCreds(t *testing.T) {
	t.Helper()
	t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
	t.Setenv("AWS_SESSION_TOKEN", "")
	// Cleared, not left ambient: region resolution is under test, so a region
	// inherited from whatever machine runs the suite would make these results
	// depend on the box. Tests that care set it back explicitly.
	t.Setenv("AWS_REGION", "")
	t.Setenv("AWS_DEFAULT_REGION", "")
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
}

// TestAcquire_ReachesEndpointOverride is the regression test for the bug this
// file exists for: with CARLOS_S3_ENDPOINT set, the lease's lock.json must be
// written through the override — not to real AWS. It asserts on the recorded
// requests, so it fails loudly (rather than hanging on a live AWS dial) if the
// leaser ever stops honouring the endpoint.
func TestAcquire_ReachesEndpointOverride(t *testing.T) {
	f, url := newFakeS3(t)
	fakeCreds(t)
	t.Setenv("CARLOS_S3_ENDPOINT", url)

	// Region deliberately empty, and fakeCreds leaves the chain empty too, so
	// this also exercises the last resort: a client signing for no region at
	// all would fail every request against the fake.
	held, err := Acquire(context.Background(), "test-bucket", "inst/example.com", "")
	if err != nil {
		t.Fatalf("Acquire against %s: %v", url, err)
	}

	got := f.requests()
	// Path-style addressing: bucket in the path, not the hostname — a
	// compatible store has no per-bucket virtual host.
	wantKey := "/test-bucket/inst/example.com/lock.json"
	var sawGet, sawPut bool
	for _, r := range got {
		switch r {
		case "GET " + wantKey:
			sawGet = true
		case "PUT " + wantKey:
			sawPut = true
		}
	}
	// An absent lock is one create-only PUT (createLease); a GET comes
	// only when a lock is present. Either way the PUT must arrive here.
	_ = sawGet
	if !sawPut {
		t.Fatalf("lease traffic did not reach the override: got %v, want a PUT of %s", got, wantKey)
	}

	// The PUT body is the lock itself — proof the acquire really happened
	// here rather than the server merely being probed.
	var lease litestream.Lease
	if err := json.Unmarshal(f.putBody(), &lease); err != nil {
		t.Fatalf("decode written lock.json (%q): %v", f.putBody(), err)
	}
	if lease.Owner == "" {
		t.Errorf("written lease carries no owner: %+v", lease)
	}

	if err := held.Release(context.Background()); err != nil {
		t.Fatalf("Release: %v", err)
	}
	var sawDelete bool
	for _, r := range f.requests() {
		if r == "DELETE "+wantKey {
			sawDelete = true
		}
	}
	if !sawDelete {
		t.Errorf("release did not reach the override: got %v", f.requests())
	}
}

// TestNewLeaserClient_EndpointConfig pins the client options either side of the
// override, because the no-override case is the flagship's production path and
// must stay byte-identical to what it was before endpoint.go existed.
func TestNewLeaserClient_EndpointConfig(t *testing.T) {
	fakeCreds(t)

	t.Run("override sets a path-style base endpoint", func(t *testing.T) {
		t.Setenv("CARLOS_S3_ENDPOINT", "https://fsn1.your-objectstorage.com")
		c, _, err := newLeaserClient(context.Background(), "eu-central-1")
		if err != nil {
			t.Fatalf("newLeaserClient: %v", err)
		}
		o := c.Options()
		if o.BaseEndpoint == nil || *o.BaseEndpoint != "https://fsn1.your-objectstorage.com" {
			t.Errorf("BaseEndpoint = %v, want the override", o.BaseEndpoint)
		}
		if !o.UsePathStyle {
			t.Error("UsePathStyle = false, want true under an endpoint override")
		}
	})

	t.Run("no override leaves stock AWS addressing", func(t *testing.T) {
		t.Setenv("CARLOS_S3_ENDPOINT", "")
		c, _, err := newLeaserClient(context.Background(), "eu-west-1")
		if err != nil {
			t.Fatalf("newLeaserClient: %v", err)
		}
		o := c.Options()
		if o.BaseEndpoint != nil {
			t.Errorf("BaseEndpoint = %q, want nil (stock AWS)", *o.BaseEndpoint)
		}
		if o.UsePathStyle {
			t.Error("UsePathStyle = true, want false without an override")
		}
	})
}

// TestNewReplicaClient_EndpointConfig covers the other half of a lease: the
// LTX files beside lock.json. A replica client that ignored the override while
// the leaser honoured it would split one lease across two stores.
func TestNewReplicaClient_EndpointConfig(t *testing.T) {
	fakeCreds(t)
	t.Run("override", func(t *testing.T) {
		t.Setenv("CARLOS_S3_ENDPOINT", "https://fsn1.your-objectstorage.com")
		c, err := newReplicaClient("b", "p", "eu-central-1")
		if err != nil {
			t.Fatal(err)
		}
		if c.Endpoint != "https://fsn1.your-objectstorage.com" {
			t.Errorf("Endpoint = %q, want the override", c.Endpoint)
		}
		if !c.ForcePathStyle {
			t.Error("ForcePathStyle = false, want true under an endpoint override")
		}
		if c.Bucket != "b" || c.Path != "p" || c.Region != "eu-central-1" {
			t.Errorf("bucket/path/region = %q/%q/%q, want b/p/eu-central-1", c.Bucket, c.Path, c.Region)
		}
	})

	t.Run("no override", func(t *testing.T) {
		t.Setenv("CARLOS_S3_ENDPOINT", "")
		c, err := newReplicaClient("b", "p", "eu-west-1")
		if err != nil {
			t.Fatal(err)
		}
		if c.Endpoint != "" || c.ForcePathStyle {
			t.Errorf("Endpoint/ForcePathStyle = %q/%v, want empty/false without an override", c.Endpoint, c.ForcePathStyle)
		}
	})
}

// TestRegionFor pins the order of authority. The case that matters most is
// "chain wins over the non-AWS default": a compatible-store deployment sets
// AWS_REGION to its own location (carlos.exciting.io uses nbg1) and that is
// what release.NewS3Store already signs for against the same bucket —
// substituting a constant would make the lease sign differently from every
// other S3 client in the process.
func TestRegionFor(t *testing.T) {
	tests := []struct {
		name, endpoint, explicit, chain, want string
	}{
		{"explicit wins over everything", "https://s3.example.com", "fsn1", "nbg1", "fsn1"},
		{"chain wins over the non-AWS default", "https://s3.example.com", "", "nbg1", "nbg1"},
		{"non-AWS default only fills a total blank", "https://s3.example.com", "", "", lss3.DefaultRegion},
		{"no override leaves a blank blank", "", "", "", ""},
		{"no override keeps the chain", "", "", "eu-west-1", "eu-west-1"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Setenv("CARLOS_S3_ENDPOINT", tt.endpoint)
			if got := regionFor(tt.explicit, tt.chain); got != tt.want {
				t.Errorf("regionFor(%q, %q) with endpoint %q = %q, want %q",
					tt.explicit, tt.chain, tt.endpoint, got, tt.want)
			}
		})
	}
}

// TestResolveRegion_PrefersAWSRegionEnv is the end-to-end version of the case
// above, through the real AWS chain: with AWS_REGION set and no explicit
// region, the lease must sign for AWS_REGION — not for litestream's non-AWS
// constant. This is the exact shape of the private CARLOS box's host.env.
func TestResolveRegion_PrefersAWSRegionEnv(t *testing.T) {
	fakeCreds(t)
	t.Setenv("CARLOS_S3_ENDPOINT", "https://nbg1.your-objectstorage.com")
	t.Setenv("AWS_REGION", "nbg1")

	got, err := resolveRegion(context.Background(), "")
	if err != nil {
		t.Fatalf("resolveRegion: %v", err)
	}
	if got != "nbg1" {
		t.Errorf("resolveRegion = %q, want %q (AWS_REGION must outrank the non-AWS default)", got, "nbg1")
	}
}

// TestRestoreCreatesTheOutputPathsParentDirectory pins the review fix: a
// nested row's instance directory does not exist before its first wake (it
// is created by the spawn path, which runs after Restore), so Restore must
// create OutputPath's parent itself rather than fail with ENOENT before the
// spawn path ever gets a chance to.
func TestRestoreCreatesTheOutputPathsParentDirectory(t *testing.T) {
	fakeCreds(t)
	// Every listing this exercises, the v0.3.x/LTX format comparison and
	// the LTX restore plan itself, is a ListObjectsV2 call; answering all
	// of them with an empty bucket makes Restore take the "no replica"
	// path, the one every fresh nested row actually takes on its first wake.
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/xml")
		fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>b</Name><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`)
	}))
	defer srv.Close()
	t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)

	root := t.TempDir()
	dbPath := filepath.Join(root, "instances", "h.example.com", "h.example.com.db")
	if _, err := os.Stat(filepath.Dir(dbPath)); !os.IsNotExist(err) {
		t.Fatalf("test setup: parent directory already exists")
	}

	restored, err := Restore(context.Background(), dbPath, "test-bucket", "inst/h.example.com", "")
	if err != nil {
		t.Fatalf("Restore into a path whose parent is absent: %v", err)
	}
	if restored {
		t.Fatal("restored = true against an empty bucket, want false (nothing to restore)")
	}
	if fi, err := os.Stat(filepath.Dir(dbPath)); err != nil || !fi.IsDir() {
		t.Fatalf("Restore did not create the output path's parent directory: %v", err)
	}
}

// ReadReplicaClient is the db door's reader (spec 2026-09-02 §6). It must
// carry the coordinates it was given and sign for the region lease.Restore
// would: an explicit region wins, and an empty one falls to the same chain
// resolution the writer uses — the endpoint override's non-AWS default when
// nothing else answers.
func TestReadReplicaClientCarriesBucketPrefixRegion(t *testing.T) {
	fakeCreds(t)
	c, err := ReadReplicaClient(context.Background(), "data-bucket", "instances/a.example.com", "eu-west-1")
	if err != nil {
		t.Fatal(err)
	}
	if c.Bucket != "data-bucket" {
		t.Errorf("Bucket = %q, want data-bucket", c.Bucket)
	}
	if c.Path != "instances/a.example.com" {
		t.Errorf("Path = %q, want instances/a.example.com", c.Path)
	}
	if c.Region != "eu-west-1" {
		t.Errorf("Region = %q, want eu-west-1", c.Region)
	}
}

func TestReadReplicaClientResolvesAnEmptyRegionLikeRestore(t *testing.T) {
	fakeCreds(t)
	t.Setenv("CARLOS_S3_ENDPOINT", "http://127.0.0.1:1")
	c, err := ReadReplicaClient(context.Background(), "data-bucket", "instances/a.example.com", "")
	if err != nil {
		t.Fatal(err)
	}
	if c.Region != lss3.DefaultRegion {
		t.Errorf("Region = %q, want litestream's non-AWS default %q (the writer's resolution)", c.Region, lss3.DefaultRegion)
	}
	if c.Endpoint != "http://127.0.0.1:1" || !c.ForcePathStyle {
		t.Errorf("Endpoint = %q ForcePathStyle = %v, want the override applied", c.Endpoint, c.ForcePathStyle)
	}
}

func TestRootReplicaClientEnvironmentKeepsDefaultChain(t *testing.T) {
	for _, region := range []string{"eu-west-1", ""} {
		t.Run("region="+region, func(t *testing.T) {
			fakeCreds(t)
			t.Setenv("AWS_SESSION_TOKEN", "test-session-token")
			t.Setenv("AWS_PROFILE", "")
			t.Setenv("AWS_DEFAULT_PROFILE", "")
			dir := t.TempDir()
			marker := filepath.Join(dir, "executed")
			config := filepath.Join(dir, "config")
			if err := os.WriteFile(config, []byte("[default]\ncredential_process = touch "+marker+"\n"), 0600); err != nil {
				t.Fatal(err)
			}
			t.Setenv("AWS_CONFIG_FILE", config)
			t.Setenv("AWS_SHARED_CREDENTIALS_FILE", config)
			var mu sync.Mutex
			var locations, listings int
			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if !strings.Contains(r.Header.Get("Authorization"), "Credential=test-key/") || r.Header.Get("X-Amz-Security-Token") != "test-session-token" {
					t.Error("request did not carry the explicit keys and session token")
				}
				w.Header().Set("Content-Type", "application/xml")
				mu.Lock()
				defer mu.Unlock()
				if r.URL.Query().Has("location") {
					locations++
					fmt.Fprint(w, `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">eu-west-1</LocationConstraint>`)
				} else {
					listings++
					fmt.Fprint(w, `<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>b</Name><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`)
				}
			}))
			defer srv.Close()
			// Keep Endpoint empty to exercise Litestream's bucket-region lookup.
			t.Setenv("CARLOS_S3_ENDPOINT", "")
			t.Setenv("AWS_ENDPOINT_URL_S3", srv.URL)
			c, err := newReplicaClientForUID("b", "p", region, 0, rootaws.OffcloudCredentials)
			if err != nil {
				t.Fatal(err)
			}
			c.ForcePathStyle = true
			if c.AccessKeyID != "" || c.SecretAccessKey != "" {
				t.Fatal("environment credentials must use the upstream default chain")
			}
			if _, err := c.GenerationsV3(context.Background()); err != nil {
				t.Fatal(err)
			}
			mu.Lock()
			defer mu.Unlock()
			if listings == 0 || (region == "" && locations != 1) || (region != "" && locations != 0) {
				t.Fatalf("locations=%d listings=%d for region %q", locations, listings, region)
			}
			if _, err := os.Stat(marker); !os.IsNotExist(err) {
				t.Fatal("credential_process executed")
			}
		})
	}
}

func TestNonRootReplicaClientKeepsDefaultProvider(t *testing.T) {
	c, err := newReplicaClientForUID("b", "p", "eu-west-1", 1001, func() (aws.Credentials, error) {
		t.Fatal("non-root read root credentials")
		return aws.Credentials{}, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if c.AccessKeyID != "" || c.SecretAccessKey != "" {
		t.Fatal("non-root credential selection changed")
	}
}

func TestRootReplicaClientOffcloudCredentials(t *testing.T) {
	for _, token := range []string{"", "FILE_TOKEN"} {
		t.Run("token="+token, func(t *testing.T) {
			body := "[default]\naws_access_key_id=FILE_ACCESS\naws_secret_access_key=FILE_SECRET\n"
			if token != "" {
				body += "aws_session_token=" + token + "\n"
			}
			c, err := newReplicaClientForUID("b", "p", "eu-west-1", 0, func() (aws.Credentials, error) {
				return rootaws.ParseStaticCredentials([]byte(body))
			})
			if token != "" {
				if !errors.Is(err, errRootReplicaSessionToken) || c != nil {
					t.Fatalf("token-bearing file: client=%v err=%v", c, err)
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if c.AccessKeyID != "FILE_ACCESS" || c.SecretAccessKey != "FILE_SECRET" {
				t.Fatal("off-cloud file keys were not set on the upstream client")
			}
		})
	}
}

func TestRootReplicaClientWithoutOffcloudKeepsDefaultChain(t *testing.T) {
	c, err := newReplicaClientForUID("b", "p", "eu-west-1", 0, func() (aws.Credentials, error) {
		return aws.Credentials{}, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if c.AccessKeyID != "" || c.SecretAccessKey != "" {
		t.Fatal("instance-profile credentials must use the upstream default chain")
	}
}

func TestRootReplicaClientRefusesUnsafeOffcloudCredentials(t *testing.T) {
	refused := errors.New("unsafe file")
	c, err := newReplicaClientForUID("b", "p", "eu-west-1", 0, func() (aws.Credentials, error) {
		return aws.Credentials{}, refused
	})
	if !errors.Is(err, refused) || c != nil {
		t.Fatalf("unsafe file: client=%v err=%v", c, err)
	}
}