refactor: update EventBus interface to support topic-based publishing and subscribing with options

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	go.uber.org/mock v0.5.0
 	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67
+	golang.org/x/sync v0.11.0
 	google.golang.org/grpc v1.70.0
 	gorm.io/gorm v1.25.12
 )

go.sum

@@ -135,6 +135,8 @@ golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 h1:1UoZQm6f0P/ZO0w1Ri+f+ifG/
 golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67/go.mod h1:qj5a5QZpwLU2NLQudwIN5koi3beDhSAlJwa67PuM98c=
 golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
 golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=

messaging/default_eventbus.go

@@ -40,19 +40,19 @@ func generateUniqueID() string {
 }
 
 // Publish sends an event to all subscribers of the specified event type asynchronously.
-func (b *MemoryEventBus) Publish(ctx context.Context, event Event) error {
+func (b *MemoryEventBus) Publish(ctx context.Context, topic string, event Event) error {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
 	if b.closed {
 		return errors.New("eventbus is closed")
 	}
 
-	if event.Type == "" || event.ID == "" {
-		return errors.New("event must have a valid ID and Type")
+	if topic == "" || event.ID == "" {
+		return errors.New("event must have a valid ID and Topic")
 	}
 
-	if chans, exists := b.subscribers[event.Type]; exists {
+	if chans, exists := b.subscribers[topic]; exists {
 		for _, sub := range chans {
 			go func(c chan EventWithCtx) {
 				select {
@@ -67,18 +67,23 @@ func (b *MemoryEventBus) Publish(ctx context.Context, event Event) error {
 
 // Subscribe registers a handler for the specified event type.
 // It returns an error if the bus is closed.
-func (b *MemoryEventBus) Subscribe(eventType string, handler func(ctx context.Context, event Event)) error {
+func (b *MemoryEventBus) Subscribe(topic string, handler func(ctx context.Context, event Event), opts ...SubscriptionOption) error {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
 	if b.closed {
 		return errors.New("eventbus is closed")
 	}
 
+	options := &SubscriptionOptions{}
+	for _, opt := range opts {
+		opt(options)
+	}
+
 	ch := make(chan EventWithCtx, 10) // Buffered channel to prevent blocking
 	id := generateUniqueID()
 	sub := subscriber{id: id, channel: ch}
-	b.subscribers[eventType] = append(b.subscribers[eventType], sub)
+	b.subscribers[topic] = append(b.subscribers[topic], sub)
 
 	go func() {
 		for e := range ch {

messaging/event.go

@@ -19,7 +19,7 @@ type Event struct {
 
 // EventBus defines the interface for publishing and subscribing to events
 type EventBus interface {
-	Publish(ctx context.Context, event Event) error
-	Subscribe(eventType string, handler func(ctx context.Context, event Event)) error
+	Publish(ctx context.Context, topic string, event Event) error
+	Subscribe(topic string, handler func(ctx context.Context, event Event), opts ...SubscriptionOption) error
 	Close() error // Clean up resources
 }

messaging/options.go

@@ -0,0 +1,24 @@
+package messaging
+
+// SubscriptionOptions holds configuration for subscription.
+type SubscriptionOptions struct {
+	Group string
+	Name  string
+}
+
+// SubscriptionOption defines a function to set subscription options.
+type SubscriptionOption func(opts *SubscriptionOptions)
+
+// WithConsumerGroup specifies the consumer group.
+func WithConsumerGroup(group string) SubscriptionOption {
+	return func(opts *SubscriptionOptions) {
+		opts.Group = group
+	}
+}
+
+// WithConsumerName specifies the consumer name.
+func WithConsumerName(name string) SubscriptionOption {
+	return func(opts *SubscriptionOptions) {
+		opts.Name = name
+	}
+}