manager

func New(opts ...Option) *Relay

{
	r := &Relay{
		broker:   broker.NewMemoryBroker(),
		handlers: make(map[string]any),
	}
	options.Apply(r, opts...)
	if r.broker == nil {
		panic("relay: broker cannot be nil")
	}
	return r
}

func Register[T any](r *Relay, topic string, fn Handler[T]) error

{
	if r == nil {
		return errors.New("relay: relay cannot be nil")
	}
	if topic == "" {
		return errors.New("relay: topic cannot be empty")
	}
	if fn == nil {
		return errors.New("relay: handler cannot be nil")
	}
	wrapper := func(ctx context.Context, raw []byte) error {
		if len(raw) > maxMessageSize {
			return errors.New("relay: message exceeds 4 MiB limit")
		}
		var payload T
		if err := json.Unmarshal(raw, &payload); err != nil {
			return fmt.Errorf("payload unmarshal failed: %w", err)
		}
		return fn(ctx, payload)
	}

	r.lifecycle.Lock()
	if r.started || r.starting || r.stopping {
		r.lifecycle.Unlock()
		return errors.New("relay: cannot register handlers after start")
	}
	r.handlerMu.Lock()
	if _, exists := r.handlers[topic]; exists {
		r.handlerMu.Unlock()
		r.lifecycle.Unlock()
		return fmt.Errorf("relay: topic %q already has a handler", topic)
	}
	r.handlers[topic] = wrapper
	r.handlerMu.Unlock()
	r.lifecycle.Unlock()
	return nil
}

func Enqueue[T any](ctx context.Context, r *Relay, topic string, payload T) error

{
	if r == nil {
		return errors.New("relay: relay cannot be nil")
	}
	data, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("payload marshal failed: %w", err)
	}
	if len(data) > maxMessageSize {
		return errors.New("relay: message exceeds 4 MiB limit")
	}

	return r.broker.Publish(ctx, topic, data)
}

func unsubscribeAll(subscriptions []broker.Subscription) error

{
	var errs []error
	for index := len(subscriptions) - 1; index >= 0; index-- {
		subscription := subscriptions[index]
		err := func() (err error) {
			defer func() {
				if recovered := recover(); recovered != nil {
					err = fmt.Errorf("relay: unsubscribe panic: %v", recovered)
				}
			}()
			return subscription.Unsubscribe()
		}()
		if err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

func TestRegisterAndEnqueue(t *testing.T)

{
	r := New()
	var mu sync.Mutex
	var received testPayload
	handlerCalled := make(chan struct{})

	Register[testPayload](r, "test.topic", func(ctx context.Context, p testPayload) error {
		mu.Lock()
		received = p
		mu.Unlock()
		close(handlerCalled)
		return nil
	})

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	ready, err := r.Start(ctx)
	if err != nil {
		t.Fatalf("Start failed: %v", err)
	}
	<-ready

	Enqueue[testPayload](context.Background(), r, "test.topic", testPayload{Message: "hello", Value: 42})

	select {
	case <-handlerCalled:
		mu.Lock()
		if received.Message != "hello" || received.Value != 42 {
			t.Errorf("got %+v, want {Message:hello Value:42}", received)
		}
		mu.Unlock()
	case <-time.After(time.Second):
		t.Fatal("timeout waiting for handler")
	}
}

func TestRegister_TypeSafety(t *testing.T)

{
	r := New()

	Register[testPayload](r, "topic1", func(ctx context.Context, p testPayload) error {
		return nil
	})
	Register[int](r, "topic2", func(ctx context.Context, p int) error {
		return nil
	})

	if len(r.handlers) != 2 {
		t.Errorf("expected 2 handlers, got %d", len(r.handlers))
	}
}

func TestStart_ContextCancel(t *testing.T)

{
	r := New()
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	ready, err := r.Start(ctx)
	if err == nil {
		t.Fatal("Start accepted a cancelled context")
	}
	<-ready
}

func TestMultipleTopics(t *testing.T)

{
	r := New()
	ch1 := make(chan int, 1)
	ch2 := make(chan int, 1)

	Register[int](r, "topic.a", func(ctx context.Context, p int) error {
		ch1 <- p
		return nil
	})
	Register[int](r, "topic.b", func(ctx context.Context, p int) error {
		ch2 <- p
		return nil
	})

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	ready, err := r.Start(ctx)
	if err != nil {
		t.Fatalf("Start failed: %v", err)
	}
	<-ready

	Enqueue[int](context.Background(), r, "topic.a", 1)
	Enqueue[int](context.Background(), r, "topic.b", 2)

	var results [2]int
	select {
	case results[0] = <-ch1:
	case <-time.After(time.Second):
		t.Fatal("timeout for topic.a")
	}
	select {
	case results[1] = <-ch2:
	case <-time.After(time.Second):
		t.Fatal("timeout for topic.b")
	}

	if results[0] != 1 || results[1] != 2 {
		t.Errorf("got %v, want [1 2]", results[:])
	}
}

func TestEnqueueAndRegisterAreThreadSafe(t *testing.T)

{
	r := New()

	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			handler := func(ctx context.Context, p int) error { return nil }
			Register[int](r, "", handler)
		}(i)
	}

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			Enqueue[int](context.Background(), r, "", 0)
		}()
	}
	wg.Wait()
}

func BenchmarkEnqueue(b *testing.B)

{
	r := New()
	Register[int](r, "bench", func(ctx context.Context, p int) error { return nil })
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go r.Start(ctx)

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		Enqueue[int](context.Background(), r, "bench", i)
	}
}

func TestStartRejectsDuplicateAndContextCancellationUnsubscribes(t *testing.T)

{
	r := New()
	var calls int
	if err := Register[int](r, "topic", func(context.Context, int) error {
		calls++
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	ready, err := r.Start(ctx)
	if err != nil {
		t.Fatal(err)
	}
	<-ready
	if _, err := r.Start(context.Background()); err == nil {
		t.Fatal("second Start() succeeded")
	}
	cancel()
	deadline := time.Now().Add(time.Second)
	for {
		r.lifecycle.Lock()
		started := r.started
		r.lifecycle.Unlock()
		if !started {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("relay did not stop after context cancellation")
		}
		time.Sleep(time.Millisecond)
	}
	if err := Enqueue(context.Background(), r, "topic", 1); err != nil {
		t.Fatal(err)
	}
	if calls != 0 {
		t.Fatalf("handler called %d times after stop", calls)
	}
}

func TestRelayBrokerCallbacksCanReenterLifecycle(t *testing.T)

{
	customBroker := &reentrantBroker{}
	relay := New(WithBroker(customBroker))
	customBroker.relay = relay
	if err := Register[int](relay, "topic", func(context.Context, int) error {
		return nil
	}); err != nil {
		t.Fatal(err)
	}

	done := make(chan error, 1)
	go func() {
		_, err := relay.Start(context.Background())
		done <- err
	}()
	select {
	case err := <-done:
		if err == nil {
			t.Fatal("Start() ignored the reentrant Stop()")
		}
	case <-time.After(time.Second):
		t.Fatal("Start() deadlocked in a reentrant broker callback")
	}
}