filter subId

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

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`})

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