ArmA-3-web-server/backend/internal/services/log_streamer_test.go

package services

import (
	"context"
	"strings"
	"sync"
	"testing"
	"time"
)

func TestLogStreamer_SubscribeUnsubscribe(t *testing.T) {
	ls := NewLogStreamer()

	ch := ls.Subscribe("server", "client1")
	if ch == nil {
		t.Fatal("Subscribe() returned nil channel")
	}

	ls.Unsubscribe("server", "client1")

	// Channel should be closed after unsubscribe
	select {
	case _, ok := <-ch:
		if ok {
			t.Error("channel should be closed after unsubscribe")
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("channel not closed within timeout")
	}
}

func TestLogStreamer_Broadcast(t *testing.T) {
	ls := NewLogStreamer()

	ch1 := ls.Subscribe("server", "client1")
	ch2 := ls.Subscribe("server", "client2")

	ls.Broadcast("server", "hello")

	for _, ch := range []chan string{ch1, ch2} {
		select {
		case line := <-ch:
			if line != "hello" {
				t.Errorf("received %q, want %q", line, "hello")
			}
		case <-time.After(100 * time.Millisecond):
			t.Error("timeout waiting for broadcast")
		}
	}
}

func TestLogStreamer_BroadcastNonBlocking(t *testing.T) {
	ls := NewLogStreamer()

	ch := ls.Subscribe("server", "client1")

	// Fill the channel (capacity 256)
	for i := 0; i < 256; i++ {
		ls.Broadcast("server", "line")
	}

	// This should not block even though channel is full
	done := make(chan struct{})
	go func() {
		ls.Broadcast("server", "dropped")
		close(done)
	}()

	select {
	case <-done:
		// Good, broadcast didn't block
	case <-time.After(100 * time.Millisecond):
		t.Error("broadcast blocked on full channel")
	}

	// Drain all queued messages
	for i := 0; i < 256; i++ {
		<-ch
	}

	// Now the next broadcast should succeed
	ls.Broadcast("server", "after")
	select {
	case line := <-ch:
		if line != "after" {
			t.Errorf("received %q, want %q", line, "after")
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("timeout waiting for broadcast")
	}
}

func TestLogStreamer_Stream(t *testing.T) {
	ls := NewLogStreamer()

	ch := ls.Subscribe("server", "client1")

	reader := strings.NewReader("line1\nline2\nline3\n")
	go ls.Stream(context.Background(), "server", reader, "DONE")

	lines := []string{}
	for i := 0; i < 4; i++ { // 3 lines + DONE close message
		select {
		case line := <-ch:
			lines = append(lines, line)
		case <-time.After(100 * time.Millisecond):
			t.Fatalf("timeout waiting for line %d", i+1)
		}
	}

	if lines[0] != "line1" || lines[1] != "line2" || lines[2] != "line3" || lines[3] != "DONE" {
		t.Errorf("lines = %v, want [line1 line2 line3 DONE]", lines)
	}
}

func TestLogStreamer_UnsubscribeCleansUpEmptyServer(t *testing.T) {
	ls := NewLogStreamer()

	ls.Subscribe("server", "client1")
	ls.Unsubscribe("server", "client1")

	// After unsubscribing the last client, the server entry should be cleaned up
	ls.mu.RLock()
	_, exists := ls.subs["server"]
	ls.mu.RUnlock()

	if exists {
		t.Error("empty server entry should be cleaned up after last unsubscribe")
	}
}

func TestLogStreamer_MultipleServers(t *testing.T) {
	ls := NewLogStreamer()

	chServer := ls.Subscribe("server", "client1")
	chSteamcmd := ls.Subscribe("steamcmd", "client1")

	ls.Broadcast("server", "server msg")
	ls.Broadcast("steamcmd", "steamcmd msg")

	select {
	case line := <-chServer:
		if line != "server msg" {
			t.Errorf("server channel received %q, want %q", line, "server msg")
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("timeout on server channel")
	}

	// SteamCMD channel should not have received the server message
	select {
	case line := <-chSteamcmd:
		if line != "steamcmd msg" {
			t.Errorf("steamcmd channel received %q, want %q", line, "steamcmd msg")
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("timeout on steamcmd channel")
	}
}

func TestLogStreamer_ConcurrentSubscribeUnsubscribe(t *testing.T) {
	ls := NewLogStreamer()
	var wg sync.WaitGroup

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			clientID := "client"
			ch := ls.Subscribe("server", clientID)
			_ = ch
			ls.Unsubscribe("server", clientID)
		}(i)
	}

	wg.Wait()
}

func TestLogStreamer_SubscribeOverwriteClosesOld(t *testing.T) {
	ls := NewLogStreamer()

	ch1 := ls.Subscribe("server", "client1")
	ls.Broadcast("server", "first")
	<-ch1

	// Re-subscribe with same clientID — old channel should be closed
	ch2 := ls.Subscribe("server", "client1")

	// Old channel should be closed
	select {
	case _, ok := <-ch1:
		if ok {
			t.Error("old channel should be closed after re-subscribe")
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("old channel not closed within timeout")
	}

	// New channel should work
	ls.Broadcast("server", "second")
	select {
	case line := <-ch2:
		if line != "second" {
			t.Errorf("new channel received %q, want %q", line, "second")
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("timeout waiting on new channel")
	}
}

func TestLogStreamer_StreamContextCancel(t *testing.T) {
	ls := NewLogStreamer()
	ch := ls.Subscribe("server", "client1")

	ctx, cancel := context.WithCancel(context.Background())
	reader := contextReader{ctx: ctx}
	go ls.Stream(ctx, "server", reader, "")

	// Broadcast should still work while stream is running
	ls.Broadcast("server", "live")
	select {
	case line := <-ch:
		if line != "live" {
			t.Errorf("received %q, want %q", line, "live")
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("timeout waiting for broadcast")
	}

	// Cancel context — stream should stop
	cancel()
	time.Sleep(100 * time.Millisecond)
}

type contextReader struct {
	ctx context.Context
}

func (r contextReader) Read(p []byte) (int, error) {
	<-r.ctx.Done()
	return 0, r.ctx.Err()
}