silo/internal/api/broker_test.go

package api

import (
	"testing"
	"time"

	"github.com/rs/zerolog"
)

func TestBrokerSubscribeUnsubscribe(t *testing.T) {
	b := NewBroker(zerolog.Nop())

	c := b.Subscribe()
	if b.ClientCount() != 1 {
		t.Fatalf("expected 1 client, got %d", b.ClientCount())
	}

	b.Unsubscribe(c)
	if b.ClientCount() != 0 {
		t.Fatalf("expected 0 clients, got %d", b.ClientCount())
	}
}

func TestBrokerPublish(t *testing.T) {
	b := NewBroker(zerolog.Nop())
	c := b.Subscribe()
	defer b.Unsubscribe(c)

	b.Publish("item.created", `{"part_number":"F01-0001"}`)

	select {
	case ev := <-c.ch:
		if ev.Type != "item.created" {
			t.Fatalf("expected type item.created, got %s", ev.Type)
		}
		if ev.ID != 1 {
			t.Fatalf("expected ID 1, got %d", ev.ID)
		}
		if ev.Data != `{"part_number":"F01-0001"}` {
			t.Fatalf("unexpected data: %s", ev.Data)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for event")
	}
}

func TestBrokerPublishDropsSlow(t *testing.T) {
	b := NewBroker(zerolog.Nop())
	c := b.Subscribe()
	defer b.Unsubscribe(c)

	// Fill the client's channel
	for i := 0; i < clientChanSize+10; i++ {
		b.Publish("heartbeat", "{}")
	}

	// Should have clientChanSize events buffered, rest dropped
	count := len(c.ch)
	if count != clientChanSize {
		t.Fatalf("expected %d buffered events, got %d", clientChanSize, count)
	}
}

func TestBrokerEventsSince(t *testing.T) {
	b := NewBroker(zerolog.Nop())

	b.Publish("item.created", `{"pn":"A"}`)
	b.Publish("item.updated", `{"pn":"B"}`)
	b.Publish("item.deleted", `{"pn":"C"}`)

	events := b.EventsSince(1) // after ID 1
	if len(events) != 2 {
		t.Fatalf("expected 2 events, got %d", len(events))
	}
	if events[0].Type != "item.updated" {
		t.Fatalf("expected item.updated, got %s", events[0].Type)
	}
	if events[1].Type != "item.deleted" {
		t.Fatalf("expected item.deleted, got %s", events[1].Type)
	}

	// No events after the latest
	events = b.EventsSince(3)
	if len(events) != 0 {
		t.Fatalf("expected 0 events, got %d", len(events))
	}
}

func TestBrokerClientCount(t *testing.T) {
	b := NewBroker(zerolog.Nop())

	c1 := b.Subscribe()
	c2 := b.Subscribe()
	c3 := b.Subscribe()

	if b.ClientCount() != 3 {
		t.Fatalf("expected 3 clients, got %d", b.ClientCount())
	}

	b.Unsubscribe(c2)
	if b.ClientCount() != 2 {
		t.Fatalf("expected 2 clients, got %d", b.ClientCount())
	}

	b.Unsubscribe(c1)
	b.Unsubscribe(c3)
	if b.ClientCount() != 0 {
		t.Fatalf("expected 0 clients, got %d", b.ClientCount())
	}
}

func TestBrokerShutdown(t *testing.T) {
	b := NewBroker(zerolog.Nop())
	c := b.Subscribe()

	b.Shutdown()

	// Client's closed channel should be closed
	select {
	case <-c.closed:
		// expected
	case <-time.After(time.Second):
		t.Fatal("client closed channel not closed after shutdown")
	}

	if b.ClientCount() != 0 {
		t.Fatalf("expected 0 clients after shutdown, got %d", b.ClientCount())
	}
}

func TestBrokerMonotonicIDs(t *testing.T) {
	b := NewBroker(zerolog.Nop())

	b.Publish("a", "{}")
	b.Publish("b", "{}")
	b.Publish("c", "{}")

	events := b.EventsSince(0)
	if len(events) != 3 {
		t.Fatalf("expected 3 events, got %d", len(events))
	}
	for i := 1; i < len(events); i++ {
		if events[i].ID <= events[i-1].ID {
			t.Fatalf("event IDs not monotonic: %d <= %d", events[i].ID, events[i-1].ID)
		}
	}
}