minstrel/internal/eventbus/eventbus_test.go

package eventbus

import (
	"sync"
	"testing"
	"time"
)

func TestPublish_DeliversToSubscriber(t *testing.T) {
	t.Parallel()
	b := New()
	ch, unsub := b.Subscribe(4)
	defer unsub()

	b.Publish(Event{Kind: "test.event", UserID: "u1", Data: map[string]any{"x": 1}})

	select {
	case got := <-ch:
		if got.Kind != "test.event" || got.UserID != "u1" {
			t.Fatalf("unexpected event: %+v", got)
		}
		if got.Data["x"] != 1 {
			t.Fatalf("payload mismatch: %+v", got.Data)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for event")
	}
}

func TestPublish_FanOutToMultipleSubscribers(t *testing.T) {
	t.Parallel()
	b := New()
	ch1, unsub1 := b.Subscribe(4)
	ch2, unsub2 := b.Subscribe(4)
	defer unsub1()
	defer unsub2()

	b.Publish(Event{Kind: "broadcast"})

	for i, ch := range []<-chan Event{ch1, ch2} {
		select {
		case got := <-ch:
			if got.Kind != "broadcast" {
				t.Fatalf("ch%d: kind=%q", i, got.Kind)
			}
		case <-time.After(time.Second):
			t.Fatalf("ch%d: timed out", i)
		}
	}
}

func TestPublish_DropsWhenSubscriberBufferFull(t *testing.T) {
	t.Parallel()
	b := New()
	_, unsub := b.Subscribe(2) // tiny buffer
	defer unsub()

	// Without draining the channel, push more events than the buffer holds.
	// Drops are silent; the assertion is that Publish does not block.
	done := make(chan struct{})
	go func() {
		for i := 0; i < 100; i++ {
			b.Publish(Event{Kind: "flood"})
		}
		close(done)
	}()

	select {
	case <-done:
		// Good — Publish returned without blocking on the laggard.
	case <-time.After(2 * time.Second):
		t.Fatal("Publish blocked on a slow subscriber")
	}
}

func TestUnsubscribe_RemovesFromBusAndClosesChannel(t *testing.T) {
	t.Parallel()
	b := New()
	ch, unsub := b.Subscribe(4)

	if got := b.SubscriberCount(); got != 1 {
		t.Fatalf("after Subscribe: SubscriberCount=%d, want 1", got)
	}

	unsub()

	if got := b.SubscriberCount(); got != 0 {
		t.Fatalf("after unsub: SubscriberCount=%d, want 0", got)
	}

	// Channel is closed: receive returns zero value with ok=false.
	if _, ok := <-ch; ok {
		t.Fatal("expected channel to be closed after unsubscribe")
	}
}

func TestUnsubscribe_Idempotent(t *testing.T) {
	t.Parallel()
	b := New()
	_, unsub := b.Subscribe(4)
	unsub()
	unsub() // must not panic / double-close
	if got := b.SubscriberCount(); got != 0 {
		t.Fatalf("SubscriberCount=%d, want 0", got)
	}
}

func TestPublish_ConcurrentWritersAndSubscribers(t *testing.T) {
	t.Parallel()
	b := New()

	const (
		subs       = 8
		perWriter  = 50
		writers    = 4
		bufferSize = 100
	)

	var subWG sync.WaitGroup
	subWG.Add(subs)
	stops := make([]func(), 0, subs)
	for i := 0; i < subs; i++ {
		ch, unsub := b.Subscribe(bufferSize)
		stops = append(stops, unsub)
		go func() {
			defer subWG.Done()
			for range ch {
				// Drain until channel closes.
			}
		}()
	}

	var pubWG sync.WaitGroup
	pubWG.Add(writers)
	for w := 0; w < writers; w++ {
		go func() {
			defer pubWG.Done()
			for i := 0; i < perWriter; i++ {
				b.Publish(Event{Kind: "concurrent"})
			}
		}()
	}
	pubWG.Wait()

	// Tear down subscribers — channels close, goroutines exit.
	for _, stop := range stops {
		stop()
	}
	subWG.Wait()
}

// Close is what lets the server shut down with SSE clients connected: each
// stream returns when its channel closes.
func TestClose_EndsEverySubscription(t *testing.T) {
	t.Parallel()
	b := New()
	ch1, unsub1 := b.Subscribe(4)
	ch2, unsub2 := b.Subscribe(4)

	b.Close()

	for i, ch := range []<-chan Event{ch1, ch2} {
		if _, ok := <-ch; ok {
			t.Fatalf("subscriber %d: channel still open after Close", i)
		}
	}
	if got := b.SubscriberCount(); got != 0 {
		t.Fatalf("SubscriberCount=%d after Close, want 0", got)
	}
	// The handlers' deferred unsubscribes still run; they must not
	// double-close.
	unsub1()
	unsub2()
	b.Close()
}

func TestSubscribe_AfterCloseIsAlreadyClosed(t *testing.T) {
	t.Parallel()
	b := New()
	b.Close()

	ch, unsub := b.Subscribe(4)
	defer unsub()
	if _, ok := <-ch; ok {
		t.Fatal("a subscription made after Close should already be closed")
	}
	if got := b.SubscriberCount(); got != 0 {
		t.Fatalf("SubscriberCount=%d, want 0", got)
	}
	b.Publish(Event{Kind: "after.close"}) // must not panic
}