package aether import ( "context" "encoding/json" "fmt" "log" "sync" "github.com/google/uuid" "github.com/nats-io/nats.go" ) // NATSEventBus is an EventBus that broadcasts events across all cluster nodes using NATS. // Supports wildcard patterns for cross-namespace subscriptions using NATS native wildcards. // // Security Considerations: // Wildcard subscriptions (using "*" or ">") receive events from multiple namespaces. // This bypasses namespace isolation at the NATS level. Ensure proper access controls // are in place at the application layer before granting wildcard subscription access. type NATSEventBus struct { *EventBus // Embed base EventBus for local subscriptions nc *nats.Conn // NATS connection subscriptions []*nats.Subscription patternSubscribers map[string]int // Track number of subscribers per pattern (includes wildcards) nodeID string // Unique ID for this node mutex sync.Mutex ctx context.Context cancel context.CancelFunc } // eventMessage is the wire format for events sent over NATS type eventMessage struct { NodeID string `json:"node_id"` NamespaceID string `json:"namespace_id"` Event *Event `json:"event"` } // NewNATSEventBus creates a new NATS-backed event bus func NewNATSEventBus(nc *nats.Conn) (*NATSEventBus, error) { ctx, cancel := context.WithCancel(context.Background()) neb := &NATSEventBus{ EventBus: NewEventBus(), nc: nc, nodeID: uuid.New().String(), subscriptions: make([]*nats.Subscription, 0), patternSubscribers: make(map[string]int), ctx: ctx, cancel: cancel, } return neb, nil } // Subscribe creates a local subscription and ensures NATS subscription exists for the pattern. // Supports NATS subject patterns: // - "*" matches a single token // - ">" matches one or more tokens (only at the end) // // Security Warning: Wildcard patterns receive events from all matching namespaces, // bypassing namespace isolation. Only use for trusted system components. func (neb *NATSEventBus) Subscribe(namespacePattern string) <-chan *Event { return neb.SubscribeWithFilter(namespacePattern, nil) } // SubscribeWithFilter creates a filtered subscription channel for a namespace pattern. // Events are filtered by the provided SubscriptionFilter before delivery. // If filter is nil or empty, all events matching the namespace pattern are delivered. // // For NATSEventBus: // - Namespace pattern filtering is applied at the NATS level using native wildcards // - EventTypes and ActorPattern filters are applied client-side after receiving messages // // This allows efficient server-side filtering for namespaces while providing // flexible client-side filtering for event types and actors. func (neb *NATSEventBus) SubscribeWithFilter(namespacePattern string, filter *SubscriptionFilter) <-chan *Event { neb.mutex.Lock() defer neb.mutex.Unlock() // Create local subscription first (with filter) ch := neb.EventBus.SubscribeWithFilter(namespacePattern, filter) // Check if this is the first subscriber for this pattern count := neb.patternSubscribers[namespacePattern] if count == 0 { // First subscriber - create NATS subscription // NATS natively supports wildcards, so we can use the pattern directly subject := fmt.Sprintf("aether.events.%s", namespacePattern) sub, err := neb.nc.Subscribe(subject, func(msg *nats.Msg) { neb.handleNATSEvent(msg, namespacePattern) }) if err != nil { log.Printf("[NATSEventBus] Failed to subscribe to NATS subject %s: %v", subject, err) // Record subscription error neb.metrics.RecordSubscribeError(namespacePattern) } else { neb.subscriptions = append(neb.subscriptions, sub) if IsWildcardPattern(namespacePattern) { log.Printf("[NATSEventBus] Node %s subscribed to wildcard pattern %s", neb.nodeID, subject) } else { log.Printf("[NATSEventBus] Node %s subscribed to %s", neb.nodeID, subject) } } } neb.patternSubscribers[namespacePattern] = count + 1 return ch } // Unsubscribe removes a local subscription and cleans up NATS subscription if no more subscribers func (neb *NATSEventBus) Unsubscribe(namespacePattern string, ch <-chan *Event) { neb.mutex.Lock() defer neb.mutex.Unlock() neb.EventBus.Unsubscribe(namespacePattern, ch) count := neb.patternSubscribers[namespacePattern] if count > 0 { count-- neb.patternSubscribers[namespacePattern] = count if count == 0 { delete(neb.patternSubscribers, namespacePattern) log.Printf("[NATSEventBus] No more subscribers for pattern %s on node %s", namespacePattern, neb.nodeID) } } } // handleNATSEvent processes events received from NATS func (neb *NATSEventBus) handleNATSEvent(msg *nats.Msg, subscribedPattern string) { var eventMsg eventMessage if err := json.Unmarshal(msg.Data, &eventMsg); err != nil { log.Printf("[NATSEventBus] Failed to unmarshal event: %v", err) return } // Skip events that originated from this node (already delivered locally) if eventMsg.NodeID == neb.nodeID { return } // For wildcard subscriptions, we need to deliver to the EventBus using // the subscribed pattern so it reaches the correct wildcard subscriber. // For exact subscriptions, use the actual namespace. if IsWildcardPattern(subscribedPattern) { // Deliver using the pattern - the EventBus will route to wildcard subscribers neb.deliverToWildcardSubscribers(subscribedPattern, eventMsg.Event) } else { // Forward to local EventBus subscribers with actual namespace neb.EventBus.Publish(eventMsg.NamespaceID, eventMsg.Event) } } // deliverToWildcardSubscribers delivers an event to subscribers of a specific wildcard pattern // Applies filters before delivery. func (neb *NATSEventBus) deliverToWildcardSubscribers(pattern string, event *Event) { neb.EventBus.mutex.RLock() defer neb.EventBus.mutex.RUnlock() for _, sub := range neb.EventBus.wildcardSubscribers { if sub.pattern == pattern { // Apply filter if present if sub.filter != nil && !sub.filter.IsEmpty() { if !sub.filter.Matches(event) { // Event doesn't match filter, skip delivery continue } } select { case sub.ch <- event: // Event delivered from NATS neb.metrics.RecordReceive(pattern) default: // Channel full, skip this subscriber (non-blocking) neb.metrics.RecordDroppedEvent(pattern) } } } } // Publish publishes an event both locally and to NATS for cross-node broadcasting func (neb *NATSEventBus) Publish(namespaceID string, event *Event) { // First publish locally neb.EventBus.Publish(namespaceID, event) // Then publish to NATS for other nodes subject := fmt.Sprintf("aether.events.%s", namespaceID) eventMsg := eventMessage{ NodeID: neb.nodeID, NamespaceID: namespaceID, Event: event, } data, err := json.Marshal(eventMsg) if err != nil { log.Printf("[NATSEventBus] Failed to marshal event for NATS: %v", err) neb.metrics.RecordPublishError(namespaceID) return } if err := neb.nc.Publish(subject, data); err != nil { log.Printf("[NATSEventBus] Failed to publish event to NATS: %v", err) neb.metrics.RecordPublishError(namespaceID) return } } // Stop closes the NATS event bus and all subscriptions func (neb *NATSEventBus) Stop() { neb.mutex.Lock() defer neb.mutex.Unlock() neb.cancel() for _, sub := range neb.subscriptions { if err := sub.Unsubscribe(); err != nil { log.Printf("[NATSEventBus] Error unsubscribing: %v", err) } } neb.subscriptions = nil neb.EventBus.Stop() log.Printf("[NATSEventBus] Node %s stopped", neb.nodeID) }