package main

import (
	"fmt"
	"sync"
	"time"
)

// Event represents a message to be published.
type Event struct {
	Topic string
	Data  string
}

// Subscriber is a channel that receives events.
type Subscriber chan Event

// EventBus manages subscriptions and event publishing.
type EventBus struct {
	subscribers map[string][]Subscriber
	mu          sync.RWMutex
}

// NewEventBus creates and returns a new EventBus.
func NewEventBus() *EventBus {
	return &EventBus{
		subscribers: make(map[string][]Subscriber),
	}
}

// Subscribe adds a new subscriber for a given topic.
// It returns a channel where the subscriber will receive events.
func (eb *EventBus) Subscribe(topic string) Subscriber {
	eb.mu.Lock()
	defer eb.mu.Unlock()

	ch := make(Subscriber, 10) // Buffered channel to avoid blocking publisher immediately
	eb.subscribers[topic] = append(eb.subscribers[topic], ch)
	fmt.Printf("Subscribed to topic '%s'.
", topic)
	return ch
}

// Unsubscribe removes a subscriber from a topic.
func (eb *EventBus) Unsubscribe(topic string, sub Subscriber) {
	eb.mu.Lock()
	defer eb.mu.Unlock()

	if subs, ok := eb.subscribers[topic]; ok {
		for i, s := range subs {
			if s == sub {
				eb.subscribers[topic] = append(subs[:i], subs[i+1:]...)
				close(s) // Close the subscriber's channel
				fmt.Printf("Unsubscribed from topic '%s'.
", topic)
				return
			}
		}
	}
}

// Publish sends an event to all subscribers of the event's topic.
func (eb *EventBus) Publish(event Event) {
	eb.mu.RLock()
	defer eb.mu.RUnlock()

	if subs, ok := eb.subscribers[event.Topic]; ok {
		fmt.Printf("Publishing event to topic '%s': %s
", event.Topic, event.Data)
		for _, sub := range subs {
			// Use a select with a default to avoid blocking the publisher
			// if a subscriber is slow to consume messages.
			select {
			case sub <- event:
				// Sent successfully
			default:
				fmt.Printf("Warning: Subscriber for topic '%s' is full, dropping event: %s
", event.Topic, event.Data)
			}
		}
	}
}

func main() {
	bus := NewEventBus()
	var wg sync.WaitGroup

	// Consumer 1 for "user_events"
	wg.Add(1)
	go func() {
		defer wg.Done()
		sub1 := bus.Subscribe("user_events")
		for event := range sub1 {
			fmt.Printf("[Consumer 1] Received event on topic '%s': %s
", event.Topic, event.Data)
		}
		fmt.Println("[Consumer 1] Stopped.")
	}()

	// Consumer 2 for "user_events" and "system_events"
	wg.Add(1)
	go func() {
		defer wg.Done()
		sub2 := bus.Subscribe("user_events")
		sub3 := bus.Subscribe("system_events")
		for {
			select {
			case event, ok := <-sub2:
				if !ok { return } // Channel closed
				fmt.Printf("[Consumer 2] User event: %s
", event.Data)
			case event, ok := <-sub3:
				if !ok { return } // Channel closed
				fmt.Printf("[Consumer 2] System event: %s
", event.Data)
			case <-time.After(5 * time.Second):
				fmt.Println("[Consumer 2] No events for 5s, unsubscribing from system_events.")
				bus.Unsubscribe("system_events", sub3)
				return // Exit this goroutine
			}
		}
	}()

	// Give consumers time to subscribe
	time.Sleep(100 * time.Millisecond)

	// Publishers
	bus.Publish(Event{Topic: "user_events", Data: "User registered: Alice"})
	time.Sleep(200 * time.Millisecond)
	bus.Publish(Event{Topic: "system_events", Data: "Database backup started"})
	time.Sleep(200 * time.Millisecond)
	bus.Publish(Event{Topic: "user_events", Data: "User logged in: Bob"})
	time.Sleep(200 * time.Millisecond)
	bus.Publish(Event{Topic: "system_events", Data: "CPU usage high"})
	time.Sleep(200 * time.Millisecond)

	// Wait for all goroutines to finish (e.g., Consumer 2 will timeout and exit)
	wg.Wait()
	fmt.Println("Main finished.")
}