diff --git a/internal/event/bus/hub.go b/internal/event/bus/hub.go index cb65e7b2..39f562a6 100644 --- a/internal/event/bus/hub.go +++ b/internal/event/bus/hub.go @@ -39,7 +39,7 @@ type Consumer struct { PID int // from Hello.ConsumerPID EventTypes []string // raw wildcard patterns from Hello Filter string // raw regex from Hello (for status display) - SubscribeID string // optional personal subscription label for status display + SubscribeID string // optional personal subscription label and local isolation key SubscribedAt time.Time SendCh chan any // bus → consume frames (Event/SourceState/Heartbeat/Bye) matcher consumerMatcher @@ -52,19 +52,22 @@ type Consumer struct { // consumerMatcher pre-compiles EventTypes wildcard patterns and the optional // Filter regex into a fast checker invoked once per delivered event per // consumer. Empty EventTypes means catch-all (everything matches except -// what Filter excludes). +// what Filter excludes). Non-empty SubscribeID is an additional exact-match +// constraint used by personal_stream consumers so same event_type subscriptions +// do not fan out to each other. type consumerMatcher struct { - catchAll bool - exact map[string]struct{} // patterns without '*' - prefixes []string // patterns ending in ".*" or "*" — store the prefix only - filter *regexp.Regexp // nil if no filter + catchAll bool + exact map[string]struct{} // patterns without '*' + prefixes []string // patterns ending in ".*" or "*" — store the prefix only + filter *regexp.Regexp // nil if no filter + subscribeID string // empty = no subscribe_id filtering } func compileMatcher(eventTypes []string, filter string, subscribeID string) (consumerMatcher, error) { m := consumerMatcher{ - exact: make(map[string]struct{}), + exact: make(map[string]struct{}), + subscribeID: strings.TrimSpace(subscribeID), } - _ = subscribeID // kept in the wire/config shape, but no longer filters fan-out. if len(eventTypes) == 0 { m.catchAll = true } @@ -102,6 +105,9 @@ func (m *consumerMatcher) matches(raw *dwsevent.RawEvent) bool { if raw == nil { return false } + if m.subscribeID != "" && raw.SubscribeID != m.subscribeID { + return false + } eventType := raw.EventType // First the include rules: catchAll OR exact-list OR prefix-list. included := m.catchAll diff --git a/internal/event/bus/hub_test.go b/internal/event/bus/hub_test.go index b46038ab..566a38ff 100644 --- a/internal/event/bus/hub_test.go +++ b/internal/event/bus/hub_test.go @@ -49,6 +49,15 @@ func drain(c *Consumer, n int, t *testing.T) []*transport.Event { return out } +func assertNoEvent(c *Consumer, t *testing.T) { + t.Helper() + select { + case f := <-c.SendCh: + t.Fatalf("unexpected frame: %#v", f) + case <-time.After(50 * time.Millisecond): + } +} + func TestHub_RegisterAssignsMonotonicID(t *testing.T) { h := NewHub(10) c1, err := h.Register(transport.Hello{ConsumerPID: 1}) @@ -110,26 +119,117 @@ func TestHub_DeliverFilterRegex(t *testing.T) { } } -func TestHub_DoesNotFilterBySubscribeID(t *testing.T) { +func TestHub_DeliverFiltersBySubscribeID(t *testing.T) { h := NewHub(10) c, err := h.Register(transport.Hello{ - EventTypes: []string{"user_im_message_receive_at"}, - SubscribeID: "sub-consumer", + EventTypes: []string{"user_im_message_receive_o2o"}, + SubscribeID: "sub-b", }) if err != nil { t.Fatal(err) } - ev := mkEvent("user_im_message_receive_at", "1") - ev.SubscribeID = "sub-upstream" + wrong := mkEvent("user_im_message_receive_o2o", "1") + wrong.SubscribeID = "sub-c" + h.Deliver(wrong) + assertNoEvent(c, t) + + right := mkEvent("user_im_message_receive_o2o", "2") + right.SubscribeID = "sub-b" + h.Deliver(right) + + got := drain(c, 1, t) + if got[0].EventID != "2" || got[0].SubscribeID != "sub-b" { + t.Fatalf("event = %#v, want sub-b event 2", got[0]) + } +} + +func TestHub_DeliverSubscribeIDSeparatesSameEventType(t *testing.T) { + h := NewHub(10) + b, err := h.Register(transport.Hello{ + EventTypes: []string{"user_im_message_receive_o2o"}, + SubscribeID: "sub-b", + }) + if err != nil { + t.Fatal(err) + } + c, err := h.Register(transport.Hello{ + EventTypes: []string{"user_im_message_receive_o2o"}, + SubscribeID: "sub-c", + }) + if err != nil { + t.Fatal(err) + } + + evB := mkEvent("user_im_message_receive_o2o", "b-msg") + evB.SubscribeID = "sub-b" + evC := mkEvent("user_im_message_receive_o2o", "c-msg") + evC.SubscribeID = "sub-c" + h.Deliver(evB) + h.Deliver(evC) + + gotB := drain(b, 1, t) + gotC := drain(c, 1, t) + if gotB[0].EventID != "b-msg" || gotB[0].SubscribeID != "sub-b" { + t.Fatalf("B consumer got %#v", gotB[0]) + } + if gotC[0].EventID != "c-msg" || gotC[0].SubscribeID != "sub-c" { + t.Fatalf("C consumer got %#v", gotC[0]) + } + assertNoEvent(b, t) + assertNoEvent(c, t) +} + +func TestHub_DeliverDropsMissingSubscribeIDForSpecificConsumer(t *testing.T) { + h := NewHub(10) + c, err := h.Register(transport.Hello{ + EventTypes: []string{"user_im_message_receive_group"}, + SubscribeID: "sub-group", + }) + if err != nil { + t.Fatal(err) + } + + h.Deliver(mkEvent("user_im_message_receive_group", "missing-sub")) + + assertNoEvent(c, t) +} + +func TestHub_DeliverEmptyConsumerSubscribeIDReceivesAnySubscribeID(t *testing.T) { + h := NewHub(10) + c, err := h.Register(transport.Hello{ + EventTypes: []string{"user_im_message_receive_o2o"}, + }) + if err != nil { + t.Fatal(err) + } + ev := mkEvent("user_im_message_receive_o2o", "1") + ev.SubscribeID = "sub-any" h.Deliver(ev) got := drain(c, 1, t) - if got[0].SubscribeID != "sub-upstream" { - t.Fatalf("event subscribe_id = %q, want raw event value", got[0].SubscribeID) + if got[0].SubscribeID != "sub-any" { + t.Fatalf("event subscribe_id = %q, want sub-any", got[0].SubscribeID) } } +func TestHub_DeliverEventTypeMismatchEvenWithSubscribeID(t *testing.T) { + h := NewHub(10) + c, err := h.Register(transport.Hello{ + EventTypes: []string{"user_im_message_receive_o2o"}, + SubscribeID: "sub-1", + }) + if err != nil { + t.Fatal(err) + } + ev := mkEvent("user_im_message_receive_group", "1") + ev.SubscribeID = "sub-1" + + h.Deliver(ev) + + assertNoEvent(c, t) +} + func TestHub_RegisterRejectsBadFilterRegex(t *testing.T) { h := NewHub(10) _, err := h.Register(transport.Hello{Filter: `(unclosed`}) diff --git a/internal/event/transport/protocol.go b/internal/event/transport/protocol.go index 28cce6c2..fdcb98bd 100644 --- a/internal/event/transport/protocol.go +++ b/internal/event/transport/protocol.go @@ -44,7 +44,7 @@ type Hello struct { ConsumerPID int `json:"consumer_pid"` EventTypes []string `json:"event_types,omitempty"` // wildcard list ("im.*", "approval.*"); empty = catch-all Filter string `json:"filter,omitempty"` // optional regex over event_type - SubscribeID string `json:"subscribe_id,omitempty"` // optional personal subscription label for status/debug + SubscribeID string `json:"subscribe_id,omitempty"` // optional personal subscription isolation key; empty = no subscribe_id filter Compact bool `json:"compact,omitempty"` // hint to status output; bus does not transform payloads // Role distinguishes a real consumer (registered for events) from an // ad-hoc tooling connection (status/list/stop). Ad-hoc connections do