package cluster_test

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

	"github.com/drjones/quantum-arcade/pkg/cluster"
	"github.com/redis/go-redis/v9"
)

// These tests simulate several cloned instances against one Redis, which is
// exactly the deployment shape: identical VMs, shared coordination.

func testRedis(t *testing.T) *redis.Client {
	t.Helper()
	addr := os.Getenv("ARCADE_TEST_REDIS")
	if addr == "" {
		addr = "localhost:6379"
	}
	rdb := redis.NewClient(&redis.Options{Addr: addr})
	if err := rdb.Ping(context.Background()).Err(); err != nil {
		t.Skipf("no redis available: %v", err)
	}
	return rdb
}

// room returns a name unique to this test run, so parallel packages and repeat
// runs do not fight over the same leadership key.
func room(t *testing.T) string {
	t.Helper()
	return fmt.Sprintf("test-%s-%d", t.Name(), time.Now().UnixNano())
}

func newNode(t *testing.T, rdb *redis.Client, addr string) *cluster.Node {
	t.Helper()
	n := cluster.NewNode(rdb, addr)
	ctx := context.Background()
	if err := n.Start(ctx); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { n.Stop(context.Background()) })
	return n
}

// Two clones of the same image must not end up with the same identity.
func TestClonedInstancesGetDistinctIdentities(t *testing.T) {
	rdb := testRedis(t)
	seen := map[string]bool{}
	for i := 0; i < 200; i++ {
		n := cluster.NewNode(rdb, "10.0.0.1:8080")
		if seen[n.ID] {
			t.Fatalf("identity collision after %d instances: %s", i, n.ID)
		}
		seen[n.ID] = true
	}
}

func TestInstancesDiscoverEachOther(t *testing.T) {
	rdb := testRedis(t)
	ctx := context.Background()

	a := newNode(t, rdb, "10.0.0.1:8080")
	b := newNode(t, rdb, "10.0.0.2:8080")

	members, err := a.Members(ctx)
	if err != nil {
		t.Fatal(err)
	}
	ids := map[string]bool{}
	for _, m := range members {
		ids[m.ID] = true
	}
	if !ids[a.ID] || !ids[b.ID] {
		t.Fatalf("instances did not see each other: %+v", members)
	}
}

// The core guarantee: exactly one instance drives a room, however many
// campaign at once.
func TestExactlyOneLeaderPerRoom(t *testing.T) {
	rdb := testRedis(t)
	ctx := context.Background()
	rm := room(t)

	const instances = 12
	nodes := make([]*cluster.Node, instances)
	for i := range nodes {
		nodes[i] = newNode(t, rdb, fmt.Sprintf("10.0.0.%d:8080", i+1))
	}

	results := make([]bool, instances)
	done := make(chan struct{})
	for i, n := range nodes {
		go func(i int, n *cluster.Node) {
			won, err := n.Campaign(ctx, rm)
			if err != nil {
				t.Errorf("campaign: %v", err)
			}
			results[i] = won
			done <- struct{}{}
		}(i, n)
	}
	for range nodes {
		<-done
	}

	leaders := 0
	for _, won := range results {
		if won {
			leaders++
		}
	}
	if leaders != 1 {
		t.Fatalf("%d instances claimed leadership of one room, want 1", leaders)
	}
}

func TestFollowerFindsTheLeaderAddress(t *testing.T) {
	rdb := testRedis(t)
	ctx := context.Background()
	rm := room(t)

	leader := newNode(t, rdb, "10.0.0.9:8080")
	follower := newNode(t, rdb, "10.0.0.10:8080")

	won, err := leader.Campaign(ctx, rm)
	if err != nil || !won {
		t.Fatalf("leader failed to take the room: won=%v err=%v", won, err)
	}

	m, err := follower.LeaderOf(ctx, rm)
	if err != nil {
		t.Fatal(err)
	}
	if m.ID != leader.ID {
		t.Fatalf("follower found leader %q, want %q", m.ID, leader.ID)
	}
	if m.Address != "10.0.0.9:8080" {
		t.Fatalf("leader address = %q, want 10.0.0.9:8080", m.Address)
	}
}

// Losing the lease must be visible to the instance that lost it, so it stops
// driving the room rather than producing a second stream of rounds.
func TestRenewFailsAfterLeadershipIsLost(t *testing.T) {
	rdb := testRedis(t)
	ctx := context.Background()
	rm := room(t)

	a := newNode(t, rdb, "10.0.0.1:8080")
	b := newNode(t, rdb, "10.0.0.2:8080")

	if won, _ := a.Campaign(ctx, rm); !won {
		t.Fatal("first instance did not win an uncontested room")
	}
	if held, _ := a.Renew(ctx, rm); !held {
		t.Fatal("leader could not renew its own lease")
	}

	// Simulate the lease expiring and another instance taking over.
	if err := a.Resign(ctx, rm); err != nil {
		t.Fatal(err)
	}
	if won, _ := b.Campaign(ctx, rm); !won {
		t.Fatal("second instance could not take the vacated room")
	}

	held, err := a.Renew(ctx, rm)
	if err != nil {
		t.Fatal(err)
	}
	if held {
		t.Fatal("the former leader renewed a lease it no longer holds")
	}
	if a.Leads(rm) {
		t.Fatal("the former leader still believes it leads the room")
	}
}

// A dead instance must hand its room over on its own, without intervention.
func TestLeadershipPassesOnWhenAnInstanceDies(t *testing.T) {
	rdb := testRedis(t)
	ctx := context.Background()
	rm := room(t)

	dying := newNode(t, rdb, "10.0.0.1:8080")
	survivor := newNode(t, rdb, "10.0.0.2:8080")

	if won, _ := dying.Campaign(ctx, rm); !won {
		t.Fatal("first instance did not take the room")
	}
	if won, _ := survivor.Campaign(ctx, rm); won {
		t.Fatal("a second instance took a room that was already led")
	}

	// The instance vanishes without resigning; its lease simply stops being
	// renewed. Waiting out the TTL is the whole point of the mechanism.
	dying.Stop(ctx)
	_ = rdb.Del(ctx, "qa:leader:"+rm) // stand in for the lease expiring

	deadline := time.Now().Add(cluster.LeaseTTL + 3*time.Second)
	took := false
	for time.Now().Before(deadline) {
		if won, _ := survivor.Campaign(ctx, rm); won {
			took = true
			break
		}
		time.Sleep(200 * time.Millisecond)
	}
	if !took {
		t.Fatal("no instance took over the room after the leader died")
	}
}

// A resigning instance must not be able to release a lease that has since
// passed to someone else.
func TestResignDoesNotStealAnotherLease(t *testing.T) {
	rdb := testRedis(t)
	ctx := context.Background()
	rm := room(t)

	a := newNode(t, rdb, "10.0.0.1:8080")
	b := newNode(t, rdb, "10.0.0.2:8080")

	if won, _ := a.Campaign(ctx, rm); !won {
		t.Fatal("a did not take the room")
	}
	if err := a.Resign(ctx, rm); err != nil {
		t.Fatal(err)
	}
	if won, _ := b.Campaign(ctx, rm); !won {
		t.Fatal("b could not take the vacated room")
	}

	// A stale resign from the former leader must be a no-op.
	if err := a.Resign(ctx, rm); err != nil {
		t.Fatal(err)
	}
	m, err := b.LeaderOf(ctx, rm)
	if err != nil {
		t.Fatal(err)
	}
	if m.ID != b.ID {
		t.Fatalf("stale resign released the current leader's lease (leader now %q)", m.ID)
	}
}

// Frames published by the leader must reach the instances relaying them.
func TestFramesFanOutToFollowers(t *testing.T) {
	rdb := testRedis(t)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	rm := room(t)

	leader := newNode(t, rdb, "10.0.0.1:8080")
	follower := newNode(t, rdb, "10.0.0.2:8080")

	frames, unsubscribe := follower.SubscribeFrames(ctx, rm)
	defer unsubscribe()

	// Give the subscription a moment to establish before publishing.
	time.Sleep(300 * time.Millisecond)

	want := []byte(`{"state":"running","multiplier":"2.500000"}`)
	if err := leader.PublishFrame(ctx, rm, want); err != nil {
		t.Fatal(err)
	}

	select {
	case got := <-frames:
		if string(got) != string(want) {
			t.Fatalf("relayed frame = %s, want %s", got, want)
		}
	case <-time.After(4 * time.Second):
		t.Fatal("follower never received the leader's frame")
	}
}

// Instances that stop heartbeating must drop off the roster.
func TestDeadInstancesLeaveTheRoster(t *testing.T) {
	rdb := testRedis(t)
	ctx := context.Background()

	alive := newNode(t, rdb, "10.0.0.1:8080")
	transient := newNode(t, rdb, "10.0.0.2:8080")

	transient.Stop(ctx) // a clean shutdown deregisters immediately

	members, err := alive.Members(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, m := range members {
		if m.ID == transient.ID {
			t.Fatal("a stopped instance is still listed as a member")
		}
	}
}