package workercluster

import (
	"context"
	"io"
	"log"
	"net"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/hashicorp/raft"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// TestClusterSmokeStats is a minimal smoke test that constructs a
// Cluster, calls Stats() before and after bootstrap, and confirms the
// reported values match what we expect. It is intentionally narrow
// (no Apply, no leader election) so it can run on every CI worker
// without the rafthttp transport overhead.
func TestClusterSmokeStats(t *testing.T) {
	addr := pickPort(t)
	dataDir := t.TempDir()
	creds := HTTPCreds{Login: "alice", Password: "secret"}

	c, err := New(&Options{
		NodeID:    "smoke-1",
		LocalAddr: addr,
		DataDir:   dataDir,
		Creds:     creds,
		Bootstrap: true,
		// Tight timeouts so the test stays sub-second even on slow CI.
		HeartbeatTimeout: 200 * time.Millisecond,
		ElectionTimeout:  600 * time.Millisecond,
		LogOutput:        io.Discard,
		Logger:           log.New(io.Discard, "", 0),
	})
	require.NoError(t, err)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	require.NoError(t, c.Start(ctx))
	t.Cleanup(func() {
		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancelShut()
		_ = c.Shutdown(shut)
	})

	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))

	stats := c.Stats()
	assert.Equal(t, "smoke-1", stats.NodeID)
	assert.Equal(t, addr, stats.LocalAddr)
	// Stats().Leader is the raft ServerAddress (host:port); the
	// webapp-friendly NodeID lookup happens in ClusterStats.
	assert.Equal(t, addr, stats.Leader,
		"Stats().Leader is the raft address of the leader")
	assert.Equal(t, raft.Leader.String(), stats.State)
	assert.NotZero(t, stats.Term, "term should advance after the first election")

	clusterStats := c.ClusterStats()
	assert.Equal(t, "smoke-1", clusterStats.NodeID)
	assert.Equal(t, "smoke-1", clusterStats.Leader,
		"ClusterStats translates the leader address back to the worker NodeID")
	assert.Contains(t, clusterStats.Voters, "smoke-1",
		"voter list must include the local node")
	assert.Equal(t, 0, clusterStats.FSMChecks, "no config.adopt applied yet")

	assert.Equal(t, "smoke-1", c.ClusterID())
	assert.Equal(t, addr, c.LocalAddr())
}

// TestClusterTestConfig_Smoke verifies the test-only helper commits a
// config.adopt entry and the FSM reflects the new version.
func TestClusterTestConfig_Smoke(t *testing.T) {
	addr := pickPort(t)
	dataDir := t.TempDir()
	creds := HTTPCreds{Login: "alice", Password: "secret"}

	c, err := New(&Options{
		NodeID:    "smoke-2",
		LocalAddr: addr,
		DataDir:   dataDir,
		Creds:     creds,
		Bootstrap: true,
		// Tight timeouts so the test stays sub-second.
		HeartbeatTimeout: 200 * time.Millisecond,
		ElectionTimeout:  600 * time.Millisecond,
		LogOutput:        io.Discard,
		Logger:           log.New(io.Discard, "", 0),
	})
	require.NoError(t, err)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	require.NoError(t, c.Start(ctx))
	t.Cleanup(func() {
		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancelShut()
		_ = c.Shutdown(shut)
	})

	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))
	require.Equal(t, raft.Leader, c.Raft().State(),
		"smoke test requires the local node to be leader")

	check := defaultTestCriticalCheck()
	idx, err := c.applyTestConfig(&check)
	require.NoError(t, err)
	assert.NotZero(t, idx, "applied index must be non-zero")

	fsmStats := c.FSM().Stats()
	assert.EqualValues(t, 1, fsmStats.ConfigVersion, "fsm should have adopted one config version")
	assert.Equal(t, 1, fsmStats.ConfigCount, "fsm should hold exactly one critical check")

	clusterStats := c.ClusterStats()
	assert.Equal(t, 1, clusterStats.FSMChecks,
		"ClusterStats().FSMChecks must mirror FSM().Stats().ConfigCount")
	assert.EqualValues(t, 1, clusterStats.FSMConfigVersion,
		"ClusterStats().FSMConfigVersion must mirror FSM().Stats().ConfigVersion")
	assert.Equal(t, 0, clusterStats.FSMOutboxLen,
		"ClusterStats().FSMOutboxLen must mirror FSM().Stats().OutboxLen (empty here)")
	assert.Equal(t, "steady", clusterStats.FSMPartition,
		"ClusterStats().FSMPartition must mirror FSM().Stats().Partition")
}

// TestClusterStats_FSMFieldsOnFreshCluster pins the FSM-side fields
// on a fresh, never-applied cluster to the documented zero values
// (ConfigVersion=0, OutboxLen=0, Partition=steady). The JSON endpoint
// relies on these being deterministic so a freshly bootstrapped cluster
// does not surprise operators with stale defaults.
func TestClusterStats_FSMFieldsOnFreshCluster(t *testing.T) {
	addr := pickPort(t)
	c, err := New(&Options{
		NodeID:           "fresh-fsm",
		LocalAddr:        addr,
		DataDir:          t.TempDir(),
		Creds:            HTTPCreds{Login: "alice", Password: "secret"},
		Bootstrap:        true,
		HeartbeatTimeout: 200 * time.Millisecond,
		ElectionTimeout:  600 * time.Millisecond,
		LogOutput:        io.Discard,
		Logger:           log.New(io.Discard, "", 0),
	})
	require.NoError(t, err)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	require.NoError(t, c.Start(ctx))
	t.Cleanup(func() {
		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancelShut()
		_ = c.Shutdown(shut)
	})
	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))

	cs := c.ClusterStats()
	assert.EqualValues(t, 0, cs.FSMConfigVersion, "no config.adopt applied yet")
	assert.Equal(t, 0, cs.FSMOutboxLen, "no outbox entries yet")
	assert.Equal(t, "steady", cs.FSMPartition,
		"newly created FSM defaults Partition.State to 'steady'")
}

// TestTestConfig_NonLeaderErrors pins the precondition that the
// helper refuses to submit an entry on a non-leader (the raft library
// would reject the apply anyway, but we want the failure to be
// deterministic and informative).
func TestTestConfig_NonLeaderErrors(t *testing.T) {
	// Three-node fixture so we have a clear "not the leader" node.
	if testing.Short() {
		t.Skip("3-node smoke skipped in -short mode")
	}
	f := newClusterFixture(t, 3)
	require.NoError(t, waitForLeader(context.Background(), f.Nodes[0], 5*time.Second))

	var leader, follower *Cluster
	for _, n := range f.Nodes {
		if n.Raft().State() == raft.Leader {
			leader = n
		} else {
			follower = n
		}
	}
	require.NotNil(t, leader)
	require.NotNil(t, follower)

	check := defaultTestCriticalCheck()
	_, err := follower.applyTestConfig(&check)
	require.Error(t, err, "non-leader must refuse test config application")
	assert.Contains(t, strings.ToLower(err.Error()), "not leader")
}

// TestClusterStats_VotersIsFreshEachCall is a tiny regression guard:
// the Voters slice returned by ClusterStats must be a fresh slice on
// every call (the Stats.Members backing store is mutated by FSM
// Apply calls). If we accidentally return the backing slice directly,
// the JSON handler would race with raft.
func TestClusterStats_VotersIsFreshEachCall(t *testing.T) {
	addr := pickPort(t)
	c, err := New(&Options{
		NodeID:           "fresh-1",
		LocalAddr:        addr,
		DataDir:          t.TempDir(),
		Creds:            HTTPCreds{Login: "alice", Password: "secret"},
		Bootstrap:        true,
		HeartbeatTimeout: 200 * time.Millisecond,
		ElectionTimeout:  600 * time.Millisecond,
		LogOutput:        io.Discard,
		Logger:           log.New(io.Discard, "", 0),
	})
	require.NoError(t, err)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	require.NoError(t, c.Start(ctx))
	t.Cleanup(func() {
		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancelShut()
		_ = c.Shutdown(shut)
	})
	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))

	s1 := c.ClusterStats()
	s2 := c.ClusterStats()
	require.NotEmpty(t, s1.Voters)
	require.NotEmpty(t, s2.Voters)

	// Mutating one must not affect the other.
	original := s1.Voters[0]
	s1.Voters[0] = "MUTATED"
	assert.Equal(t, original, s2.Voters[0],
		"Voters slices must not share backing storage")
}

// keep atomic referenced so the import isn't flagged on minimal edits.
var _ atomic.Int32

// ensureLocalAddrIsLoopback is a build-tag helper that the demo and
// other helpers can call to confirm we never accidentally bind a
// public address.
func ensureLocalAddrIsLoopback(addr string) error {
	host, _, err := net.SplitHostPort(addr)
	if err != nil {
		return err
	}
	if !strings.HasPrefix(host, "127.") && host != "::1" {
		return &loopbackErr{host: host}
	}
	return nil
}

type loopbackErr struct{ host string }

func (e *loopbackErr) Error() string { return "non-loopback host: " + e.host }