package events import ( "strings" "sync" "time" ) // Bus is a labeled publish/subscribe event bus. // Handlers subscribe to patterns (exact or trailing wildcard). // Publishing fans out to all matching subscribers. type Bus struct { mu sync.RWMutex subs map[string][]*subscription seq uint64 // subscription ID counter } type subscription struct { id uint64 pattern string handler Handler } // NewBus creates a new event bus. func NewBus() *Bus { return &Bus{ subs: make(map[string][]*subscription), } } // Subscribe registers a handler for a label pattern. // Returns an unsubscribe function. // // Patterns: // // "chat.message.abc123" — exact match // "chat.message.*" — wildcard: matches chat.message.{anything} // "chat.*" — wildcard: matches chat.{anything} // "*" — matches all events func (b *Bus) Subscribe(pattern string, handler Handler) func() { b.mu.Lock() b.seq++ sub := &subscription{id: b.seq, pattern: pattern, handler: handler} b.subs[pattern] = append(b.subs[pattern], sub) b.mu.Unlock() return func() { b.mu.Lock() defer b.mu.Unlock() subs := b.subs[pattern] for i, s := range subs { if s.id == sub.id { b.subs[pattern] = append(subs[:i], subs[i+1:]...) break } } } } // Publish dispatches an event to all matching subscribers. // Handlers are called synchronously in subscription order. // Use PublishAsync for non-blocking dispatch. func (b *Bus) Publish(event Event) { b.mu.RLock() var matched []Handler for pattern, subs := range b.subs { if match(event.Label, pattern) { for _, s := range subs { matched = append(matched, s.handler) } } } b.mu.RUnlock() for _, h := range matched { h(event) } } // PublishAsync dispatches an event to all matching subscribers // in separate goroutines. Useful for I/O-heavy handlers. func (b *Bus) PublishAsync(event Event) { b.mu.RLock() var matched []Handler for pattern, subs := range b.subs { if match(event.Label, pattern) { for _, s := range subs { matched = append(matched, s.handler) } } } b.mu.RUnlock() for _, h := range matched { go h(event) } } // match checks if a concrete label matches a subscription pattern. // // "chat.message.abc" matches "chat.message.abc" (exact) // "chat.message.abc" matches "chat.message.*" (wildcard) // "chat.message.abc" matches "chat.*" (wildcard) // "chat.message.abc" matches "*" (global wildcard) func match(label, pattern string) bool { if pattern == "*" { return true } if pattern == label { return true } if !strings.HasSuffix(pattern, "*") { return false } prefix := pattern[:len(pattern)-1] // "chat.message." from "chat.message.*" return strings.HasPrefix(label, prefix) } // WaitFor subscribes to an exact label and blocks until an event arrives // or the timeout elapses. Returns the event and true, or a zero Event and // false on timeout. The subscription is automatically removed after. // // Used by the browser tool bridge: the server publishes tool.call.{id} // to the client and then calls WaitFor("tool.result.{id}", 30s) to block // until the browser returns the result. func (b *Bus) WaitFor(label string, timeout time.Duration) (Event, bool) { ch := make(chan Event, 1) unsub := b.Subscribe(label, func(e Event) { select { case ch <- e: default: } }) defer unsub() select { case e := <-ch: return e, true case <-time.After(timeout): return Event{}, false } }