RegionIO/internal/network/streamer_test.go

package network

import (
	"context"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"regionio/internal/protocol"
	"regionio/internal/world"
)

// TestSpiralOrderCenterFirst confirms the centre coordinate is returned first
// and the ring order expands outward (Chebyshev distance non-decreasing).
func TestSpiralOrderCenterFirst(t *testing.T) {
	order := spiralOrder(0, 0, 2)
	if order[0] != [2]int32{0, 0} {
		t.Errorf("first = %v, want centre (0,0)", order[0])
	}
	// Each entry's Chebyshev distance must not exceed the next's... actually it
	// only needs to be non-decreasing within ring blocks. Verify the max
	// distance of the first k entries grows as expected by checking the full
	// set contains exactly the (2*2+1)² = 49 distinct coords.
	want := (2*2 + 1) * (2*2 + 1)
	if len(order) != want {
		t.Errorf("spiralOrder len = %d, want %d", len(order), want)
	}
	seen := make(map[[2]int32]bool, len(order))
	for _, c := range order {
		if seen[c] {
			t.Errorf("duplicate %v in spiral order", c)
		}
		seen[c] = true
	}
}

// TestSpiralOrderRingStructure checks that all distance-0 coords come before
// distance-1, which come before distance-2 (centre-outward ordering).
func TestSpiralOrderRingStructure(t *testing.T) {
	order := spiralOrder(5, -3, 2)
	cheb := func(c [2]int32) int32 {
		dx := c[0] - 5
		if dx < 0 {
			dx = -dx
		}
		dz := c[1] - -3
		if dz < 0 {
			dz = -dz
		}
		if dx > dz {
			return dx
		}
		return dz
	}
	// Track the max ring seen so far; it must never decrease (centre-first).
	var maxRing int32
	for _, c := range order {
		r := cheb(c)
		if r < maxRing {
			t.Errorf("ring %d appeared after ring %d — not centre-first", r, maxRing)
		}
		if r > maxRing {
			maxRing = r
		}
	}
	if maxRing != 2 {
		t.Errorf("max ring = %d, want 2", maxRing)
	}
}

// TestRequestRecenterNonBlocking confirms requestRecenter never blocks the
// caller even when many requests are pushed rapidly (the streamer drains stale
// ones). This is the property the read loop relies on to stay responsive.
func TestRequestRecenterNonBlocking(t *testing.T) {
	s := newStreamer(nil, nil, nil, 4)
	done := make(chan struct{})
	go func() {
		for i := 0; i < 1000; i++ {
			s.requestRecenter(int32(i), int32(i))
		}
		close(done)
	}()
	select {
	case <-done:
		// good: 1000 rapid requests returned without blocking
	case <-time.After(2 * time.Second):
		t.Fatal("requestRecenter blocked for 2s")
	}
}

func TestStreamerSendsStrictlyNearFirst(t *testing.T) {
	cache := world.NewCache(-1, func(cx, cz int32) *world.Chunk {
		return world.NewChunk(cx, cz, world.BiomePlains)
	})
	recorder := &recordingConn{}
	s := newStreamer(cache, NewConn(recorder), nil, 2)
	s.genRadius = s.viewRadius
	s.poolSize = 4
	if _, superseded := s.processRecenter(context.Background(), 7, -3); superseded {
		t.Fatal("unexpected recenter supersession")
	}
	defer s.tickets.Close()

	lastDistance := int32(-1)
	chunks := 0
	for _, packet := range recorder.take(t) {
		if packet.ID != protocol.PlayLevelChunk {
			continue
		}
		r := packet.Body()
		x, err := r.Int32()
		if err != nil {
			t.Fatal(err)
		}
		z, err := r.Int32()
		if err != nil {
			t.Fatal(err)
		}
		distance := chunkDistanceFrom(7, -3, [2]int32{x, z})
		if distance < lastDistance {
			t.Fatalf("chunk (%d,%d) at distance %d arrived after distance %d", x, z, distance, lastDistance)
		}
		lastDistance = distance
		chunks++
	}
	if chunks != 25 {
		t.Fatalf("level chunks sent = %d, want 25", chunks)
	}
	if got := cache.Stats().Tickets; got != 25 {
		t.Fatalf("tickets = %d, want 25", got)
	}
}

func TestStreamerQueuedRecenterStopsOldOuterRings(t *testing.T) {
	cache := world.NewCache(-1, func(cx, cz int32) *world.Chunk {
		return world.NewChunk(cx, cz, world.BiomePlains)
	})
	s := newStreamer(cache, nil, nil, 2)
	s.genRadius = s.viewRadius
	s.poolSize = 4
	s.requestRecenter(100, 100)
	next, superseded := s.processRecenter(context.Background(), 0, 0)
	defer s.tickets.Close()
	if !superseded || next != (recenterReq{cx: 100, cz: 100}) {
		t.Fatalf("superseded=%v next=%+v, want latest (100,100)", superseded, next)
	}
	if len(s.loaded) != 1 || !s.loaded[[2]int32{0, 0}] {
		t.Fatalf("old loaded set = %v, want only old center", s.loaded)
	}
}

func TestStreamerRecenterForgetsViewAndReplacesPrefetchTickets(t *testing.T) {
	cache := world.NewCacheWithLimit(-1, func(cx, cz int32) *world.Chunk {
		return world.NewChunk(cx, cz, world.BiomePlains)
	}, nil, 9)
	recorder := &recordingConn{}
	s := newStreamer(cache, NewConn(recorder), nil, 2)
	s.viewRadius, s.genRadius, s.poolSize = 0, 1, 4
	defer s.tickets.Close()
	if _, superseded := s.processRecenter(context.Background(), 0, 0); superseded {
		t.Fatal("unexpected first recenter supersession")
	}
	if len(s.loaded) != 1 || cache.Stats().Tickets != 9 {
		t.Fatalf("first lifecycle loaded=%v stats=%+v", s.loaded, cache.Stats())
	}
	recorder.take(t)

	if _, superseded := s.processRecenter(context.Background(), 10, 10); superseded {
		t.Fatal("unexpected second recenter supersession")
	}
	if len(s.loaded) != 1 || !s.loaded[[2]int32{10, 10}] {
		t.Fatalf("second loaded set = %v, want only (10,10)", s.loaded)
	}
	stats := cache.Stats()
	if stats.Tickets != 9 || stats.Chunks > 9 || hasLevelChunkPacket(recorder.take(t), protocol.PlayForgetLevelChunk) != 1 {
		t.Fatalf("second lifecycle stats=%+v; want 9 tickets, <=9 chunks and one forget", stats)
	}
}

func hasLevelChunkPacket(packets []protocol.Packet, id int32) int {
	count := 0
	for _, packet := range packets {
		if packet.ID == id {
			count++
		}
	}
	return count
}

func TestLoadSixteenClientStreamersBoundedAndReleasesTickets(t *testing.T) {
	var active, peak atomic.Int32
	gen := func(cx, cz int32) *world.Chunk {
		now := active.Add(1)
		for {
			old := peak.Load()
			if now <= old || peak.CompareAndSwap(old, now) {
				break
			}
		}
		time.Sleep(200 * time.Microsecond)
		active.Add(-1)
		return world.NewChunk(cx, cz, world.BiomePlains)
	}
	cache := world.NewCacheWithLimit(-1, gen, nil, 32)
	ctx, cancel := context.WithCancel(context.Background())
	var wg sync.WaitGroup
	const clients = 16
	const groups = 4
	recorders := make([]*recordingConn, clients)
	for i := 0; i < clients; i++ {
		recorders[i] = &recordingConn{}
		s := newStreamer(cache, NewConn(recorders[i]), nil, 2)
		// A 3x3 view produces 144 ticket claims. Four spawn regions exercise
		// both shared tickets and concurrent independent generation.
		s.viewRadius, s.genRadius, s.poolSize = 1, 1, 4
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			group := int32(index % groups)
			s.requestRecenter(group*64, group*64)
			s.run(ctx)
		}(i)
	}

	deadline := time.Now().Add(120 * time.Second)
	for (cache.Stats().Frames < groups*9 || cache.Stats().Tickets < clients*9 || clientsWithPacket(recorders, protocol.PlayLevelChunk) < clients) && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}
	stats := cache.Stats()
	if stats.Frames < groups*9 || stats.Tickets != clients*9 || clientsWithPacket(recorders, protocol.PlayLevelChunk) != clients {
		cancel()
		wg.Wait()
		t.Fatalf("loaded stats = %+v, want at least %d frames and %d tickets", stats, groups*9, clients*9)
	}
	if got := peak.Load(); got > 8 {
		cancel()
		wg.Wait()
		t.Fatalf("peak concurrent generators = %d, want <= shared limit 8", got)
	}

	cancel()
	wg.Wait()
	deadline = time.Now().Add(5 * time.Second)
	for cache.Stats().Tickets != 0 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if stats = cache.Stats(); stats.Tickets != 0 || stats.Chunks > 32 {
		t.Fatalf("after disconnect stats = %+v, want zero tickets and <=32 chunks", stats)
	}
}

func clientsWithPacket(recorders []*recordingConn, id int32) int {
	count := 0
	for _, recorder := range recorders {
		if recorder.countPacketID(id) > 0 {
			count++
		}
	}
	return count
}