core/server/events/bus_test.go

package events

import (
	"encoding/json"
	"sync/atomic"
	"testing"
	"time"
)

func TestMatch(t *testing.T) {
	tests := []struct {
		label, pattern string
		want           bool
	}{
		{"workflow.assigned.abc", "workflow.assigned.abc", true},
		{"workflow.assigned.abc", "workflow.assigned.*", true},
		{"workflow.assigned.abc", "workflow.*", true},
		{"workflow.assigned.abc", "*", true},
		{"workflow.assigned.abc", "workflow.assigned.xyz", false},
		{"workflow.assigned.abc", "notification.new.*", false},
		{"workflow.assigned.abc", "workflow.assigned", false},
		{"ping", "ping", true},
		{"ping", "pong", false},
		{"plugin.hook.pre_completion", "plugin.hook.*", true},
		{"plugin.hook.pre_completion", "plugin.*", true},
	}

	for _, tt := range tests {
		got := match(tt.label, tt.pattern)
		if got != tt.want {
			t.Errorf("match(%q, %q) = %v, want %v", tt.label, tt.pattern, got, tt.want)
		}
	}
}

func TestBusPublishExact(t *testing.T) {
	bus := NewBus()
	var count int32

	bus.Subscribe("workflow.assigned.abc", func(e Event) {
		atomic.AddInt32(&count, 1)
	})

	bus.Publish(Event{Label: "workflow.assigned.abc", Payload: json.RawMessage(`{}`)})
	bus.Publish(Event{Label: "workflow.assigned.xyz", Payload: json.RawMessage(`{}`)})

	if atomic.LoadInt32(&count) != 1 {
		t.Errorf("expected 1 dispatch, got %d", count)
	}
}

func TestBusPublishWildcard(t *testing.T) {
	bus := NewBus()
	var count int32

	bus.Subscribe("workflow.assigned.*", func(e Event) {
		atomic.AddInt32(&count, 1)
	})

	bus.Publish(Event{Label: "workflow.assigned.abc", Payload: json.RawMessage(`{}`)})
	bus.Publish(Event{Label: "workflow.assigned.xyz", Payload: json.RawMessage(`{}`)})
	bus.Publish(Event{Label: "workflow.claimed.abc", Payload: json.RawMessage(`{}`)})

	if atomic.LoadInt32(&count) != 2 {
		t.Errorf("expected 2 dispatches, got %d", count)
	}
}

func TestBusUnsubscribe(t *testing.T) {
	bus := NewBus()
	var count int32

	unsub := bus.Subscribe("test.event", func(e Event) {
		atomic.AddInt32(&count, 1)
	})

	bus.Publish(Event{Label: "test.event", Payload: json.RawMessage(`{}`)})
	unsub()
	bus.Publish(Event{Label: "test.event", Payload: json.RawMessage(`{}`)})

	if atomic.LoadInt32(&count) != 1 {
		t.Errorf("expected 1 dispatch after unsub, got %d", count)
	}
}

func TestBusGlobalWildcard(t *testing.T) {
	bus := NewBus()
	var count int32

	bus.Subscribe("*", func(e Event) {
		atomic.AddInt32(&count, 1)
	})

	bus.Publish(Event{Label: "workflow.assigned.abc", Payload: json.RawMessage(`{}`)})
	bus.Publish(Event{Label: "system.notify", Payload: json.RawMessage(`{}`)})
	bus.Publish(Event{Label: "plugin.hook.pre_completion", Payload: json.RawMessage(`{}`)})

	if atomic.LoadInt32(&count) != 3 {
		t.Errorf("expected 3 dispatches, got %d", count)
	}
}

func TestRouteFor(t *testing.T) {
	tests := []struct {
		label string
		want  Direction
	}{
		{"system.notify", DirToClient},
		{"plugin.hook.pre_completion", DirLocal},
		{"internal.db.write", DirLocal},
		{"unknown.event", DirLocal}, // default
		{"ping", DirFromClient},
		{"pong", DirToClient},
		// Tool bridge routes
		{"tool.call.abc123", DirToClient},
		{"tool.result.abc123", DirBoth}, // v0.32.0: DirBoth for cross-pod WaitFor
		// Extension lifecycle
		{"extension.loaded", DirLocal},
		{"extension.error", DirLocal},
		// Realtime (v0.5.0)
		{"realtime.chat.message", DirToClient},
		{"realtime.custom.event", DirToClient},
		// Room management (v0.5.0)
		{"room.subscribe", DirFromClient},
		{"room.unsubscribe", DirFromClient},
	}

	for _, tt := range tests {
		got := RouteFor(tt.label)
		if got != tt.want {
			t.Errorf("RouteFor(%q) = %d, want %d", tt.label, got, tt.want)
		}
	}
}

func TestWaitFor_Success(t *testing.T) {
	bus := NewBus()

	// Publish after a short delay
	go func() {
		time.Sleep(10 * time.Millisecond)
		bus.Publish(Event{
			Label:   "tool.result.abc123",
			Payload: json.RawMessage(`{"result":"hello"}`),
		})
	}()

	event, ok := bus.WaitFor("tool.result.abc123", 1*time.Second)
	if !ok {
		t.Fatal("WaitFor timed out, expected success")
	}
	if event.Label != "tool.result.abc123" {
		t.Errorf("expected label tool.result.abc123, got %s", event.Label)
	}
	if string(event.Payload) != `{"result":"hello"}` {
		t.Errorf("unexpected payload: %s", string(event.Payload))
	}
}

func TestWaitFor_Timeout(t *testing.T) {
	bus := NewBus()

	_, ok := bus.WaitFor("tool.result.never", 50*time.Millisecond)
	if ok {
		t.Error("WaitFor should have timed out")
	}
}

func TestWaitFor_IgnoresOtherLabels(t *testing.T) {
	bus := NewBus()

	go func() {
		time.Sleep(5 * time.Millisecond)
		bus.Publish(Event{Label: "tool.result.other", Payload: json.RawMessage(`{}`)})
		time.Sleep(5 * time.Millisecond)
		bus.Publish(Event{Label: "tool.result.target", Payload: json.RawMessage(`{"ok":true}`)})
	}()

	event, ok := bus.WaitFor("tool.result.target", 1*time.Second)
	if !ok {
		t.Fatal("WaitFor timed out")
	}
	if event.Label != "tool.result.target" {
		t.Errorf("got wrong label: %s", event.Label)
	}
}

func TestWaitFor_CleansUpSubscription(t *testing.T) {
	bus := NewBus()

	// WaitFor with immediate timeout
	bus.WaitFor("tool.result.cleanup", 1*time.Millisecond)
	time.Sleep(5 * time.Millisecond)

	// Verify no panic when publishing to the label after cleanup
	bus.Publish(Event{Label: "tool.result.cleanup", Payload: json.RawMessage(`{}`)})
}

func TestRealtimeRouting(t *testing.T) {
	// realtime.* events should be sent to clients but not accepted from them
	if !ShouldSendToClient("realtime.chat.message") {
		t.Error("realtime.* should be sent to client")
	}
	if ShouldAcceptFromClient("realtime.chat.message") {
		t.Error("realtime.* should NOT be accepted from client")
	}

	// room.subscribe/unsubscribe should be accepted from clients
	if !ShouldAcceptFromClient("room.subscribe") {
		t.Error("room.subscribe should be accepted from client")
	}
	if !ShouldAcceptFromClient("room.unsubscribe") {
		t.Error("room.unsubscribe should be accepted from client")
	}
	if ShouldSendToClient("room.subscribe") {
		t.Error("room.subscribe should NOT be sent to client")
	}
}

func TestToolCallRouteToClient(t *testing.T) {
	if !ShouldSendToClient("tool.call.abc123") {
		t.Error("tool.call.* should be sent to client")
	}
	if ShouldAcceptFromClient("tool.call.abc123") {
		t.Error("tool.call.* should NOT be accepted from client")
	}
}

func TestToolResultRouteBoth(t *testing.T) {
	// v0.32.0: tool.result is DirBoth so results cross pods for WaitFor.
	// The WS subscriber explicitly filters out tool.result events to
	// prevent re-sending to clients (see subscribeToBus in ws.go).
	if !ShouldAcceptFromClient("tool.result.abc123") {
		t.Error("tool.result.* should be accepted from client")
	}
	if !ShouldSendToClient("tool.result.abc123") {
		t.Error("tool.result.* should route DirBoth (filtered by WS subscriber, not routing table)")
	}
}