// Ported from lchat internal/activator/replica_test.go.
//
// SPIKE tests (issue #550 Part 4 v2). Live S3 tests are gated on
// CARLOS_TEST_S3_BUCKET; without it everything skips so `go test ./...`
// stays green on CI and dev machines with no AWS credentials.
package lease

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

	_ "modernc.org/sqlite"
)

func liveConfig(t *testing.T) (bucket, prefix, region string) {
	t.Helper()
	bucket = os.Getenv("CARLOS_TEST_S3_BUCKET")
	if bucket == "" {
		t.Skip("CARLOS_TEST_S3_BUCKET not set; skipping live S3 spike test")
	}
	region = os.Getenv("CARLOS_TEST_S3_REGION")
	if region == "" {
		region = "eu-west-1"
	}
	b := make([]byte, 4)
	_, _ = rand.Read(b)
	prefix = fmt.Sprintf("spike/%d-%s", time.Now().Unix(), hex.EncodeToString(b))
	t.Cleanup(func() { cleanupPrefix(t, bucket, prefix, region) })
	return bucket, prefix, region
}

// seedDB creates a WAL-mode SQLite database via modernc.org/sqlite (the
// driver this repo already uses — no cgo anywhere in the test).
func seedDB(t *testing.T, path string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatalf("open sqlite: %v", err)
	}
	defer db.Close()
	if _, err := db.Exec(`PRAGMA journal_mode=WAL`); err != nil {
		t.Fatalf("wal mode: %v", err)
	}
	if _, err := db.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil {
		t.Fatalf("create table: %v", err)
	}
	for i := 0; i < 50; i++ {
		if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("row-%d", i)); err != nil {
			t.Fatalf("insert: %v", err)
		}
	}
}

// s3Client is the verification client the live tests list and clean up with.
// It deliberately goes through the package's OWN constructor rather than
// building its own: a helper that dialled stock AWS while the code under test
// honoured CARLOS_S3_ENDPOINT would fail every assertion against a compatible
// store and strand the objects it was meant to delete.
func s3Client(t *testing.T, region string) *awss3.Client {
	t.Helper()
	client, _, err := newLeaserClient(context.Background(), region)
	if err != nil {
		t.Fatalf("aws config: %v", err)
	}
	return client
}

func listKeys(t *testing.T, bucket, prefix, region string) []string {
	t.Helper()
	client := s3Client(t, region)
	var keys []string
	p := awss3.NewListObjectsV2Paginator(client, &awss3.ListObjectsV2Input{
		Bucket: aws.String(bucket),
		Prefix: aws.String(prefix),
	})
	for p.HasMorePages() {
		out, err := p.NextPage(context.Background())
		if err != nil {
			t.Fatalf("list objects: %v", err)
		}
		for _, o := range out.Contents {
			keys = append(keys, *o.Key)
		}
	}
	return keys
}

// cleanupPrefix deletes every object AND version under the test prefix so
// spike runs leave nothing behind in the bucket.
func cleanupPrefix(t *testing.T, bucket, prefix, region string) {
	t.Helper()
	client := s3Client(t, region)
	ctx := context.Background()
	p := awss3.NewListObjectVersionsPaginator(client, &awss3.ListObjectVersionsInput{
		Bucket: aws.String(bucket),
		Prefix: aws.String(prefix),
	})
	for p.HasMorePages() {
		out, err := p.NextPage(ctx)
		if err != nil {
			t.Logf("cleanup list versions: %v", err)
			return
		}
		var objs []s3types.ObjectIdentifier
		for _, v := range out.Versions {
			objs = append(objs, s3types.ObjectIdentifier{Key: v.Key, VersionId: v.VersionId})
		}
		for _, m := range out.DeleteMarkers {
			objs = append(objs, s3types.ObjectIdentifier{Key: m.Key, VersionId: m.VersionId})
		}
		if len(objs) == 0 {
			continue
		}
		if _, err := client.DeleteObjects(ctx, &awss3.DeleteObjectsInput{
			Bucket: aws.String(bucket),
			Delete: &s3types.Delete{Objects: objs},
		}); err != nil {
			t.Logf("cleanup delete: %v", err)
		}
	}
}

// TestHeldError_ErrorsAs is a unit test (no AWS): the package's contract is
// that a contested lease surfaces as *HeldError through errors.As, carrying
// the holder and expiry.
func TestHeldError_ErrorsAs(t *testing.T) {
	exp := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC)
	wrapped := fmt.Errorf("acquire lease: %w", &HeldError{Owner: "peer-1", ExpiresAt: exp})
	var held *HeldError
	if !errors.As(wrapped, &held) {
		t.Fatalf("errors.As failed to find *HeldError in %v", wrapped)
	}
	if held.Owner != "peer-1" || !held.ExpiresAt.Equal(exp) {
		t.Fatalf("HeldError fields lost in wrap: %+v", held)
	}
	if msg := held.Error(); !strings.Contains(msg, "peer-1") {
		t.Fatalf("Error() should name the holder: %q", msg)
	}
}

// TestReplicateWithLease_ReplicatesToS3 proves Q1+Q2: the embedded library
// (pure Go, modernc driver) replicates a real database to S3 under a lease.
func TestReplicateWithLease_ReplicatesToS3(t *testing.T) {
	bucket, prefix, region := liveConfig(t)
	ctx := context.Background()

	dbPath := filepath.Join(t.TempDir(), "spike.db")
	seedDB(t, dbPath)

	r, err := ReplicateWithLease(ctx, dbPath, bucket, prefix, region)
	if err != nil {
		t.Fatalf("ReplicateWithLease: %v", err)
	}
	defer r.Close(ctx)

	if err := r.SyncAndWait(ctx); err != nil {
		t.Fatalf("SyncAndWait: %v", err)
	}

	keys := listKeys(t, bucket, prefix, region)
	t.Logf("objects under %s: %v", prefix, keys)
	var haveLock, haveLTX bool
	for _, k := range keys {
		if strings.HasSuffix(k, "lock.json") {
			haveLock = true
		}
		if strings.Contains(k, ".ltx") {
			haveLTX = true
		}
	}
	if !haveLock {
		t.Errorf("expected %s/lock.json in bucket, keys=%v", prefix, keys)
	}
	if !haveLTX {
		t.Errorf("expected replicated ltx data under %s, keys=%v", prefix, keys)
	}

	if err := r.Close(ctx); err != nil {
		t.Fatalf("Close: %v", err)
	}
}

// TestReplicateWithLease_MutualExclusion proves Q3: a second claimant on the
// same replica path fails LOUDLY at lease acquisition with a typed error —
// unlike the CLI, which silently absorbs the second writer.
func TestReplicateWithLease_MutualExclusion(t *testing.T) {
	bucket, prefix, region := liveConfig(t)
	ctx := context.Background()

	dir := t.TempDir()
	db1 := filepath.Join(dir, "one.db")
	db2 := filepath.Join(dir, "two.db")
	seedDB(t, db1)
	seedDB(t, db2)

	first, err := ReplicateWithLease(ctx, db1, bucket, prefix, region)
	if err != nil {
		t.Fatalf("first claimant: %v", err)
	}
	defer first.Close(ctx)

	// Second claimant, same bucket+prefix: must fail, and fail typed.
	second, err := ReplicateWithLease(ctx, db2, bucket, prefix, region)
	if err == nil {
		second.Close(ctx)
		t.Fatal("second claimant acquired the lease while the first held it — mutual exclusion broken")
	}
	var lexists *HeldError
	if !errors.As(err, &lexists) {
		t.Fatalf("second claimant failed, but not with *HeldError: %v", err)
	}
	t.Logf("second claimant failed loudly (typed): %v", err)
	t.Logf("  holder=%q expires=%s", lexists.Owner, lexists.ExpiresAt.Format(time.RFC3339))

	// Release, then the next claimant must succeed immediately (no TTL wait).
	if err := first.Close(ctx); err != nil {
		t.Fatalf("release: %v", err)
	}
	third, err := ReplicateWithLease(ctx, db2, bucket, prefix, region)
	if err != nil {
		t.Fatalf("claimant after release should succeed: %v", err)
	}
	if err := third.Close(ctx); err != nil {
		t.Fatalf("third close: %v", err)
	}
}

// TestDarkHostCannotReclaimAfterFailover proves the dark-then-returns half of
// failover (issue #53): host A holds the lease and goes DARK — stops renewing
// but never releases, the kill -9 / power-cut shape. Its lock.json stays in
// S3 with the expiry already stamped on it, so:
//
//  1. while that expiry is still in the future a peer is refused — dark is
//     not the same as expired, and nobody may jump a live lease;
//  2. once it lapses, host B claims it — the failover wake;
//  3. A comes back and tries to pick up where it left off, and MUST lose,
//     loudly, with a *HeldError. A returning dark host silently reclaiming a
//     lease that has moved on is the split-brain this lock exists to prevent.
//
// The other half of failover — a fallback host's activator waking the
// instance from the regional bucket — is the existing activator wake path and
// is covered by the activator's own tests; nothing about it is new here.
//
// Runs on a 15s TTL through the AcquireTTL seam; at the production LeaseTTL
// the same sequence would take 15 minutes.
func TestDarkHostCannotReclaimAfterFailover(t *testing.T) {
	bucket, prefix, region := liveConfig(t)
	ctx := context.Background()
	const shortTTL = 15 * time.Second

	a, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL)
	if err != nil {
		t.Fatalf("A initial acquire: %v", err)
	}
	acquiredAt := time.Now()
	// A goes dark. rn.stop() cancels the renewal goroutine's context and
	// waits for it to exit; it makes no S3 call of any kind, so lock.json
	// survives untouched and simply runs out its stamped TTL. Without this
	// line A would renew every shortTTL/3 and step 2 below could never
	// happen — that is what makes the stop load-bearing.
	a.rn.stop()

	// Dark is not expired: A's lock is still in force, so B must be refused —
	// and refused BY THE LEASE, not by a throttle, a 5xx or a credential
	// hiccup, which would let this leg pass while proving nothing. So the
	// refusal has to be a *HeldError carrying a live expiry, exactly as
	// TestReplicateWithLease_MutualExclusion demands of its second claimant.
	// (Both "hosts" run in this one process and therefore share the leaser's
	// owner string, so the expiry — not the name — is what identifies the
	// holder here.)
	peer, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL)
	if err == nil {
		elapsed := time.Since(acquiredAt)
		_ = peer.Release(ctx)
		if elapsed >= shortTTL {
			t.Fatalf("second acquire only returned %s after A acquired, past the %s TTL: this environment is too slow for the test's timing margins, which is not the same as a mutual-exclusion failure", elapsed, shortTTL)
		}
		t.Fatalf("B acquired after %s while A's un-expired %s lease was still held", elapsed, shortTTL)
	}
	var stillHeld *HeldError
	if !errors.As(err, &stillHeld) {
		t.Fatalf("peer refused while A was dark-but-live, but not with *HeldError: %v — an S3 or credential failure here would make this leg vacuous", err)
	}
	if !stillHeld.ExpiresAt.After(time.Now()) {
		t.Fatalf("peer refused by an ALREADY-EXPIRED lease (expires %s, %s after A acquired); want A's still-live lease", stillHeld.ExpiresAt.Format(time.RFC3339), time.Since(acquiredAt))
	}
	t.Logf("dark is not expired: peer refused by A's live lease until %s", stillHeld.ExpiresAt.Format(time.RFC3339))

	time.Sleep(shortTTL + 5*time.Second) // let A's abandoned lease lapse

	b, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL)
	if err != nil {
		t.Fatalf("B acquire after A's lease expired (the failover wake): %v", err)
	}
	bHeldFrom := time.Now()
	defer b.Release(ctx)

	// A returns from the dead and tries to resume.
	if revenant, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL); err == nil {
		elapsed := time.Since(bHeldFrom)
		_ = revenant.Release(ctx)
		if elapsed >= shortTTL {
			// B's own lease had lapsed by then (B is not renewed here either),
			// so A winning proves nothing. Say so rather than cry split brain.
			t.Fatalf("A's re-acquire only returned %s after B took over, past the %s TTL: this environment is too slow for the test's timing margins", elapsed, shortTTL)
		}
		t.Fatal("returning dark host reclaimed the lease B now holds — split brain")
	} else {
		var held *HeldError
		if !errors.As(err, &held) {
			t.Fatalf("returning dark host got %v, want *HeldError (it must lose the race)", err)
		}
		// The refusal has to come from B's LIVE lease. An expiry already in
		// the past would mean the acquire lost for some other reason and the
		// test would be proving nothing.
		if !held.ExpiresAt.After(time.Now()) {
			t.Fatalf("returning host refused by an already-expired lease (expires %s); want B's live lease", held.ExpiresAt.Format(time.RFC3339))
		}
		t.Logf("returning dark host lost the race, as required: %v", err)
	}
}