package actor

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"google.golang.org/protobuf/proto"
)

type testState struct {
	Value int `json:"value"`
}

type testGrain struct {
	id       uint64
	accessed time.Time
}

func (g *testGrain) GetId() uint64                      { return g.id }
func (g *testGrain) GetLastAccess() time.Time           { return g.accessed }
func (g *testGrain) GetLastChange() time.Time           { return g.accessed }
func (g *testGrain) GetCurrentState() (*testState, error) {
	return &testState{Value: int(g.id)}, nil
}

type mockHost struct {
	name        string
	healthy     bool
	closed      atomic.Bool
	closeCalls  atomic.Int32
	actorIDs    []uint64
	negotiateFn func([]string) ([]string, error)
	mu          sync.Mutex
	ownership   []uint64
	expiry      []uint64
}

func newMockHost(name string) *mockHost {
	return &mockHost{name: name, healthy: true}
}

func (m *mockHost) AnnounceExpiry(ids []uint64) {
	m.mu.Lock()
	m.expiry = append(m.expiry, ids...)
	m.mu.Unlock()
}

func (m *mockHost) Negotiate(otherHosts []string) ([]string, error) {
	if m.negotiateFn != nil {
		return m.negotiateFn(otherHosts)
	}
	return otherHosts, nil
}

func (m *mockHost) Name() string { return m.name }

func (m *mockHost) Proxy(_ uint64, _ http.ResponseWriter, _ *http.Request, _ io.Reader) (bool, error) {
	return false, fmt.Errorf("not implemented")
}

func (m *mockHost) Apply(_ context.Context, id uint64, _ ...proto.Message) (*MutationResult[testState], error) {
	return &MutationResult[testState]{Result: testState{Value: int(id)}}, nil
}

func (m *mockHost) Get(_ context.Context, id uint64) (*testState, error) {
	return &testState{Value: int(id)}, nil
}

func (m *mockHost) GetActorIds() []uint64 { return m.actorIDs }

func (m *mockHost) Close() error {
	m.closeCalls.Add(1)
	m.closed.Store(true)
	return nil
}

func (m *mockHost) Ping() bool { return m.healthy }

func (m *mockHost) IsHealthy() bool { return m.healthy }

func (m *mockHost) AnnounceOwnership(_ string, ids []uint64) {
	m.mu.Lock()
	m.ownership = append(m.ownership, ids...)
	m.mu.Unlock()
}

func newTestPool(t *testing.T, spawnHost func(string) (Host[testState], error)) *SimpleGrainPool[testState] {
	t.Helper()
	reg := NewMutationRegistry()
	pool, err := NewSimpleGrainPool(GrainPoolConfig[testState]{
		Hostname: "local",
		Spawn: func(_ context.Context, id uint64) (Grain[testState], error) {
			return &testGrain{id: id, accessed: time.Now()}, nil
		},
		SpawnHost:        spawnHost,
		TTL:              time.Hour,
		PoolSize:         100,
		MutationRegistry: reg,
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(pool.Close)
	return pool
}

func TestSimpleGrainPoolGetAndApply(t *testing.T) {
	pool := newTestPool(t, func(string) (Host[testState], error) {
		return nil, fmt.Errorf("no remotes")
	})

	got, err := pool.Get(context.Background(), 42)
	if err != nil {
		t.Fatal(err)
	}
	if got.Value != 42 {
		t.Fatalf("Get value = %d, want 42", got.Value)
	}

	ids := pool.GetLocalIds()
	if len(ids) != 1 || ids[0] != 42 {
		t.Fatalf("GetLocalIds = %v, want [42]", ids)
	}
}

func TestSimpleGrainPoolAddRemoteDedup(t *testing.T) {
	var spawnCalls atomic.Int32
	host := newMockHost("peer-a")

	pool := newTestPool(t, func(name string) (Host[testState], error) {
		spawnCalls.Add(1)
		if name != host.name {
			t.Fatalf("spawn host %q, want %q", name, host.name)
		}
		return host, nil
	})

	const workers = 8
	var wg sync.WaitGroup
	wg.Add(workers)
	for range workers {
		go func() {
			defer wg.Done()
			if _, err := pool.AddRemote(host.name); err != nil {
				t.Error(err)
			}
		}()
	}
	wg.Wait()

	if pool.RemoteCount() != 1 {
		t.Fatalf("RemoteCount = %d, want 1", pool.RemoteCount())
	}
	if spawnCalls.Load() != 1 {
		t.Fatalf("spawn calls = %d, want 1", spawnCalls.Load())
	}
	if host.closeCalls.Load() != 0 {
		t.Fatalf("duplicate remote closed = %d, want 0", host.closeCalls.Load())
	}
}

func TestSimpleGrainPoolAddRemoteRaceClosesDuplicate(t *testing.T) {
	start := make(chan struct{})
	var spawnCalls atomic.Int32
	var spawned []*mockHost

	pool := newTestPool(t, func(name string) (Host[testState], error) {
		spawnCalls.Add(1)
		h := newMockHost(name)
		spawned = append(spawned, h)
		if spawnCalls.Load() == 1 {
			<-start
		}
		return h, nil
	})

	go func() {
		if _, err := pool.AddRemote("peer-b"); err != nil {
			t.Error(err)
		}
	}()
	time.Sleep(20 * time.Millisecond)
	close(start)
	if _, err := pool.AddRemote("peer-b"); err != nil {
		t.Fatal(err)
	}

	if pool.RemoteCount() != 1 {
		t.Fatalf("RemoteCount = %d, want 1", pool.RemoteCount())
	}
	if spawnCalls.Load() != 2 {
		t.Fatalf("spawn calls = %d, want 2", spawnCalls.Load())
	}

	closed := 0
	deadline := time.After(2 * time.Second)
	for closed < 1 {
		closed = 0
		for _, h := range spawned {
			if h.closeCalls.Load() > 0 {
				closed++
			}
		}
		if closed >= 1 {
			break
		}
		select {
		case <-deadline:
			t.Fatal("timed out waiting for duplicate remote host Close")
		default:
			time.Sleep(10 * time.Millisecond)
		}
	}
}

func TestSimpleGrainPoolRemoveHost(t *testing.T) {
	host := newMockHost("peer-c")
	pool := newTestPool(t, func(string) (Host[testState], error) { return host, nil })

	if _, err := pool.AddRemote(host.name); err != nil {
		t.Fatal(err)
	}
	pool.remoteMu.Lock()
	pool.remoteOwners[99] = host
	pool.remoteMu.Unlock()

	pool.RemoveHost(host.name)

	if pool.RemoteCount() != 0 {
		t.Fatalf("RemoteCount = %d, want 0", pool.RemoteCount())
	}
	if owner, ok := pool.OwnerHost(99); ok || owner != nil {
		t.Fatalf("OwnerHost still mapped: %#v ok=%v", owner, ok)
	}
	if host.closeCalls.Load() != 1 {
		t.Fatalf("Close calls = %d, want 1", host.closeCalls.Load())
	}
}

func TestSimpleGrainPoolHandleOwnershipChange(t *testing.T) {
	host := newMockHost("peer-d")
	pool := newTestPool(t, func(string) (Host[testState], error) { return host, nil })

	pool.localMu.Lock()
	pool.grains[7] = &testGrain{id: 7, accessed: time.Now()}
	pool.localMu.Unlock()

	if err := pool.HandleOwnershipChange(host.name, []uint64{7}); err != nil {
		t.Fatal(err)
	}

	pool.localMu.RLock()
	_, local := pool.grains[7]
	pool.localMu.RUnlock()
	if local {
		t.Fatal("local grain should be removed after ownership change")
	}

	owner, ok := pool.OwnerHost(7)
	if !ok || owner.Name() != host.name {
		t.Fatalf("OwnerHost = (%v, %v), want (%q, true)", owner, ok, host.name)
	}
}

func TestSimpleGrainPoolPurgeEvictsStaleGrains(t *testing.T) {
	pool := newTestPool(t, func(string) (Host[testState], error) {
		return nil, fmt.Errorf("no remotes")
	})
	pool.ttl = time.Minute

	stale := &testGrain{id: 1, accessed: time.Now().Add(-2 * time.Minute)}
	fresh := &testGrain{id: 2, accessed: time.Now()}

	pool.localMu.Lock()
	pool.grains[1] = stale
	pool.grains[2] = fresh
	pool.localMu.Unlock()

	pool.purge()

	pool.localMu.RLock()
	defer pool.localMu.RUnlock()
	if _, ok := pool.grains[1]; ok {
		t.Fatal("stale grain should be purged")
	}
	if _, ok := pool.grains[2]; !ok {
		t.Fatal("fresh grain should remain")
	}
}

func TestSimpleGrainPoolBroadcastOwnership(t *testing.T) {
	host := newMockHost("peer-e")
	pool := newTestPool(t, func(string) (Host[testState], error) { return host, nil })
	if _, err := pool.AddRemote(host.name); err != nil {
		t.Fatal(err)
	}

	pool.broadcastOwnership([]uint64{11, 12})

	host.mu.Lock()
	defer host.mu.Unlock()
	if len(host.ownership) != 2 {
		t.Fatalf("ownership announces = %v, want 2 ids", host.ownership)
	}
}