uopi/internal/broker/broker_test.go

package broker_test

import (
	"context"
	"log/slog"
	"testing"
	"time"

	"github.com/uopi/uopi/internal/broker"
	"github.com/uopi/uopi/internal/datasource/stub"
)

func newBroker(tb testing.TB) (*broker.Broker, context.CancelFunc) {
	tb.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	log := slog.Default()
	b := broker.New(ctx, log)
	ds := stub.New()
	if err := ds.Connect(ctx); err != nil {
		tb.Fatal(err)
	}
	b.Register(ds)
	return b, cancel
}

func recv(t *testing.T, ch <-chan broker.Update, timeout time.Duration) broker.Update {
	t.Helper()
	select {
	case u := <-ch:
		return u
	case <-time.After(timeout):
		t.Fatalf("timed out waiting for update after %s", timeout)
		return broker.Update{}
	}
}

func TestFanOut(t *testing.T) {
	b, cancel := newBroker(t)
	defer cancel()

	ref := broker.SignalRef{DS: "stub", Name: "sine_1hz"}
	ch1 := make(chan broker.Update, 10)
	ch2 := make(chan broker.Update, 10)

	unsub1, err := b.Subscribe(ref, ch1)
	if err != nil {
		t.Fatal(err)
	}
	unsub2, err := b.Subscribe(ref, ch2)
	if err != nil {
		t.Fatal(err)
	}
	defer unsub2()

	// Both channels must receive an update
	recv(t, ch1, 2*time.Second)
	recv(t, ch2, 2*time.Second)

	// Only one upstream subscription should exist
	if n := b.ActiveSubscriptions(); n != 1 {
		t.Errorf("expected 1 active subscription, got %d", n)
	}

	// After unsubscribing ch1, ch2 should still receive updates
	unsub1()
	// drain any buffered updates
	for len(ch2) > 0 {
		<-ch2
	}
	recv(t, ch2, 2*time.Second)
}

func TestUnsubscribeLastClientTearsDownUpstream(t *testing.T) {
	b, cancel := newBroker(t)
	defer cancel()

	ref := broker.SignalRef{DS: "stub", Name: "ramp_10s"}
	ch := make(chan broker.Update, 10)

	unsub, err := b.Subscribe(ref, ch)
	if err != nil {
		t.Fatal(err)
	}
	recv(t, ch, 2*time.Second)

	if n := b.ActiveSubscriptions(); n != 1 {
		t.Errorf("expected 1 active subscription before unsub, got %d", n)
	}

	unsub()
	// Give the broker a moment to process the teardown
	time.Sleep(50 * time.Millisecond)

	if n := b.ActiveSubscriptions(); n != 0 {
		t.Errorf("expected 0 active subscriptions after unsub, got %d", n)
	}
}

func TestUnknownDataSource(t *testing.T) {
	b, cancel := newBroker(t)
	defer cancel()

	ref := broker.SignalRef{DS: "nonexistent", Name: "signal"}
	ch := make(chan broker.Update, 1)

	_, err := b.Subscribe(ref, ch)
	if err == nil {
		t.Fatal("expected error for unknown data source, got nil")
	}
}

func TestUnknownSignal(t *testing.T) {
	b, cancel := newBroker(t)
	defer cancel()

	ref := broker.SignalRef{DS: "stub", Name: "does_not_exist"}
	ch := make(chan broker.Update, 1)

	_, err := b.Subscribe(ref, ch)
	if err == nil {
		t.Fatal("expected error for unknown signal, got nil")
	}
}

func TestMaxUpdateRate(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	const maxHz = 5.0 // 5 Hz → 200 ms interval

	ds := stub.New() // counter_fast fires at 1 kHz
	if err := ds.Connect(ctx); err != nil {
		t.Fatal(err)
	}
	b := broker.New(ctx, slog.Default(), broker.WithMaxUpdateRate(maxHz))
	b.Register(ds)

	ref := broker.SignalRef{DS: "stub", Name: "counter_fast"}
	ch := make(chan broker.Update, 512)
	unsub, err := b.Subscribe(ref, ch)
	if err != nil {
		t.Fatal(err)
	}
	defer unsub()

	// Collect updates for 2 seconds.
	var count int
	deadline := time.After(2 * time.Second)
loop:
	for {
		select {
		case <-ch:
			count++
		case <-deadline:
			break loop
		}
	}

	// At 5 Hz for 2 s we expect ~10 updates (±3 for timer jitter + coalescing).
	t.Logf("MaxUpdateRate: received %d updates at %.0f Hz limit over 2s (expect ~10)", count, maxHz)
	if count < 5 || count > 20 {
		t.Errorf("expected ~10 updates at 5 Hz, got %d", count)
	}
}

func TestMaxUpdateRate_UnlimitedByDefault(t *testing.T) {
	b, cancel := newBroker(t) // no WithMaxUpdateRate → unlimited
	defer cancel()

	ref := broker.SignalRef{DS: "stub", Name: "counter_fast"}
	ch := make(chan broker.Update, 4096)
	unsub, err := b.Subscribe(ref, ch)
	if err != nil {
		t.Fatal(err)
	}
	defer unsub()

	// counter_fast fires at 1 kHz; over 500 ms we should see ≥ 400 updates.
	time.Sleep(500 * time.Millisecond)
	got := len(ch)
	t.Logf("Unlimited: received %d updates in 500ms from 1kHz signal", got)
	if got < 400 {
		t.Errorf("expected >= 400 updates with no rate limit, got %d", got)
	}
}

func TestMultipleSignals(t *testing.T) {
	b, cancel := newBroker(t)
	defer cancel()

	signals := []string{"sine_1hz", "sine_01hz", "ramp_10s", "toggle_1hz"}
	channels := make([]chan broker.Update, len(signals))
	unsubs := make([]func(), len(signals))

	for i, name := range signals {
		ch := make(chan broker.Update, 10)
		channels[i] = ch
		unsub, err := b.Subscribe(broker.SignalRef{DS: "stub", Name: name}, ch)
		if err != nil {
			t.Fatalf("subscribe %s: %v", name, err)
		}
		unsubs[i] = unsub
	}
	defer func() {
		for _, u := range unsubs {
			u()
		}
	}()

	if n := b.ActiveSubscriptions(); n != len(signals) {
		t.Errorf("expected %d active subscriptions, got %d", len(signals), n)
	}

	for i, ch := range channels {
		recv(t, ch, 2*time.Second)
		_ = i
	}
}