aether/eventbus.go

package aether

import (
	"context"
	"sync"
)

// EventBroadcaster defines the interface for publishing and subscribing to events.
//
// Subscribe accepts namespace patterns following NATS subject matching conventions:
//   - Exact match: "tenant-a" matches only "tenant-a"
//   - Single wildcard: "*" matches any single token, "tenant-*" matches "tenant-a", "tenant-b"
//   - Multi-token wildcard: ">" matches one or more tokens (only at end of pattern)
//
// Security Warning: Wildcard subscriptions bypass namespace isolation.
// Only grant wildcard access to trusted system components.
type EventBroadcaster interface {
	// Subscribe creates a channel that receives events matching the namespace pattern.
	// Pattern syntax follows NATS conventions: "*" matches single token, ">" matches multiple.
	Subscribe(namespacePattern string) <-chan *Event
	Unsubscribe(namespacePattern string, ch <-chan *Event)
	Publish(namespaceID string, event *Event)
	Stop()
	SubscriberCount(namespaceID string) int
}

// MetricsProvider is an optional interface that EventBroadcaster implementations
// can implement to expose metrics.
type MetricsProvider interface {
	// Metrics returns the metrics collector for this broadcaster.
	Metrics() BroadcasterMetrics
}

// subscription represents a single subscriber channel with its pattern
type subscription struct {
	pattern string
	ch      chan *Event
}

// EventBus broadcasts events to multiple subscribers within a namespace.
// Supports wildcard patterns for cross-namespace subscriptions.
//
// Security Considerations:
// Wildcard subscriptions (using "*" or ">") receive events from multiple namespaces.
// This is intentional for cross-cutting concerns like logging, monitoring, and auditing.
// However, it bypasses namespace isolation - use with appropriate access controls.
type EventBus struct {
	// exactSubscribers holds subscribers for exact namespace matches (no wildcards)
	exactSubscribers map[string][]chan *Event
	// wildcardSubscribers holds subscribers with wildcard patterns
	wildcardSubscribers []subscription
	mutex               sync.RWMutex
	ctx                 context.Context
	cancel              context.CancelFunc
	metrics             *DefaultMetricsCollector
}

// NewEventBus creates a new event bus
func NewEventBus() *EventBus {
	ctx, cancel := context.WithCancel(context.Background())
	return &EventBus{
		exactSubscribers:    make(map[string][]chan *Event),
		wildcardSubscribers: make([]subscription, 0),
		ctx:                 ctx,
		cancel:              cancel,
		metrics:             NewMetricsCollector(),
	}
}

// Metrics returns the metrics collector for this event bus.
func (eb *EventBus) Metrics() BroadcasterMetrics {
	return eb.metrics
}

// Subscribe creates a new subscription channel for a namespace pattern.
// Patterns follow NATS subject matching conventions:
//   - "*" matches a single token (any sequence without ".")
//   - ">" matches one or more tokens (only valid at the end)
//   - Exact strings match exactly
//
// Security Warning: Wildcard patterns receive events from all matching namespaces,
// bypassing namespace isolation. Only use for trusted system components.
func (eb *EventBus) Subscribe(namespacePattern string) <-chan *Event {
	eb.mutex.Lock()
	defer eb.mutex.Unlock()

	// Create buffered channel to prevent blocking publishers
	ch := make(chan *Event, 100)

	if IsWildcardPattern(namespacePattern) {
		// Store wildcard subscription separately
		eb.wildcardSubscribers = append(eb.wildcardSubscribers, subscription{
			pattern: namespacePattern,
			ch:      ch,
		})
	} else {
		// Exact match subscription
		eb.exactSubscribers[namespacePattern] = append(eb.exactSubscribers[namespacePattern], ch)
	}

	// Record subscription metric
	eb.metrics.RecordSubscribe(namespacePattern)

	return ch
}

// Unsubscribe removes a subscription channel
func (eb *EventBus) Unsubscribe(namespacePattern string, ch <-chan *Event) {
	eb.mutex.Lock()
	defer eb.mutex.Unlock()

	if IsWildcardPattern(namespacePattern) {
		// Remove from wildcard subscribers
		for i, sub := range eb.wildcardSubscribers {
			if sub.ch == ch {
				eb.wildcardSubscribers = append(eb.wildcardSubscribers[:i], eb.wildcardSubscribers[i+1:]...)
				close(sub.ch)
				// Record unsubscription metric
				eb.metrics.RecordUnsubscribe(namespacePattern)
				break
			}
		}
	} else {
		// Remove from exact subscribers
		subs := eb.exactSubscribers[namespacePattern]
		for i, subscriber := range subs {
			if subscriber == ch {
				// Remove channel from slice
				eb.exactSubscribers[namespacePattern] = append(subs[:i], subs[i+1:]...)
				close(subscriber)
				// Record unsubscription metric
				eb.metrics.RecordUnsubscribe(namespacePattern)
				break
			}
		}

		// Clean up empty namespace entries
		if len(eb.exactSubscribers[namespacePattern]) == 0 {
			delete(eb.exactSubscribers, namespacePattern)
		}
	}
}

// Publish sends an event to all subscribers of a namespace.
// Events are delivered to:
//   - All exact subscribers for the namespace
//   - All wildcard subscribers whose pattern matches the namespace
func (eb *EventBus) Publish(namespaceID string, event *Event) {
	eb.mutex.RLock()
	defer eb.mutex.RUnlock()

	// Record publish metric
	eb.metrics.RecordPublish(namespaceID)

	// Deliver to exact subscribers
	subscribers := eb.exactSubscribers[namespaceID]
	for _, ch := range subscribers {
		select {
		case ch <- event:
			// Event delivered
			eb.metrics.RecordReceive(namespaceID)
		default:
			// Channel full, skip this subscriber (non-blocking)
			eb.metrics.RecordDroppedEvent(namespaceID)
		}
	}

	// Deliver to matching wildcard subscribers
	for _, sub := range eb.wildcardSubscribers {
		if MatchNamespacePattern(sub.pattern, namespaceID) {
			select {
			case sub.ch <- event:
				// Event delivered
				eb.metrics.RecordReceive(namespaceID)
			default:
				// Channel full, skip this subscriber (non-blocking)
				eb.metrics.RecordDroppedEvent(namespaceID)
			}
		}
	}
}

// Stop closes the event bus
func (eb *EventBus) Stop() {
	eb.mutex.Lock()
	defer eb.mutex.Unlock()

	eb.cancel()

	// Close all exact subscriber channels and update metrics
	for namespaceID, subs := range eb.exactSubscribers {
		for _, ch := range subs {
			close(ch)
			eb.metrics.RecordUnsubscribe(namespaceID)
		}
	}

	// Close all wildcard subscriber channels and update metrics
	for _, sub := range eb.wildcardSubscribers {
		close(sub.ch)
		eb.metrics.RecordUnsubscribe(sub.pattern)
	}

	eb.exactSubscribers = make(map[string][]chan *Event)
	eb.wildcardSubscribers = make([]subscription, 0)
}

// SubscriberCount returns the number of subscribers for a namespace.
// This counts only exact match subscribers, not wildcard subscribers that may match.
func (eb *EventBus) SubscriberCount(namespaceID string) int {
	eb.mutex.RLock()
	defer eb.mutex.RUnlock()
	return len(eb.exactSubscribers[namespaceID])
}

// WildcardSubscriberCount returns the number of wildcard subscribers.
// These are subscribers using "*" or ">" patterns that may receive events
// from multiple namespaces.
func (eb *EventBus) WildcardSubscriberCount() int {
	eb.mutex.RLock()
	defer eb.mutex.RUnlock()
	return len(eb.wildcardSubscribers)
}