From 6789a8dee027762815d80d8afaf2b24dd2bb73ce Mon Sep 17 00:00:00 2001
From: mlogclub <mlog1@qq.com>
Date: Tue, 21 Apr 2026 16:42:29 +0800
Subject: [PATCH] feat(eventbus): refactor handler storage to use a slice and
 add tests for handler order

---
 internal/pkg/eventbus/bus.go      | 33 ++++++++++++++-----
 internal/pkg/eventbus/bus_test.go | 55 +++++++++++++++++++++++++++++++
 2 files changed, 79 insertions(+), 9 deletions(-)

diff --git a/internal/pkg/eventbus/bus.go b/internal/pkg/eventbus/bus.go
index f576c7f..2a97fec 100644
--- a/internal/pkg/eventbus/bus.go
+++ b/internal/pkg/eventbus/bus.go
@@ -14,9 +14,14 @@ type Handler[T any] func(ctx context.Context, event T) error
 
 type ErrorHandler func(ctx context.Context, err error)
 
+type handlerEntry[T any] struct {
+	id      uint64
+	handler Handler[T]
+}
+
 type Bus[T any] struct {
 	mu       sync.RWMutex
-	handlers map[uint64]Handler[T]
+	handlers []handlerEntry[T]
 	nextID   uint64
 
 	onError ErrorHandler
@@ -26,9 +31,7 @@ type Bus[T any] struct {
 }
 
 func New[T any](opts ...Option[T]) *Bus[T] {
-	b := &Bus[T]{
-		handlers: make(map[uint64]Handler[T]),
-	}
+	b := &Bus[T]{}
 	for _, opt := range opts {
 		opt(b)
 	}
@@ -60,7 +63,10 @@ func (b *Bus[T]) Subscribe(h Handler[T]) (uint64, func()) {
 	id := atomic.AddUint64(&b.nextID, 1)
 
 	b.mu.Lock()
-	b.handlers[id] = h
+	b.handlers = append(b.handlers, handlerEntry[T]{
+		id:      id,
+		handler: h,
+	})
 	b.mu.Unlock()
 
 	return id, func() {
@@ -87,7 +93,10 @@ func (b *Bus[T]) SubscribeOnce(h Handler[T]) (uint64, func()) {
 	id = atomic.AddUint64(&b.nextID, 1)
 
 	b.mu.Lock()
-	b.handlers[id] = wrapper
+	b.handlers = append(b.handlers, handlerEntry[T]{
+		id:      id,
+		handler: wrapper,
+	})
 	b.mu.Unlock()
 
 	return id, func() {
@@ -97,7 +106,13 @@ func (b *Bus[T]) SubscribeOnce(h Handler[T]) (uint64, func()) {
 
 func (b *Bus[T]) Unsubscribe(id uint64) {
 	b.mu.Lock()
-	delete(b.handlers, id)
+	for i, entry := range b.handlers {
+		if entry.id != id {
+			continue
+		}
+		b.handlers = append(b.handlers[:i], b.handlers[i+1:]...)
+		break
+	}
 	b.mu.Unlock()
 }
 
@@ -130,8 +145,8 @@ func (b *Bus[T]) snapshotHandlers() []Handler[T] {
 	defer b.mu.RUnlock()
 
 	handlers := make([]Handler[T], 0, len(b.handlers))
-	for _, h := range b.handlers {
-		handlers = append(handlers, h)
+	for _, entry := range b.handlers {
+		handlers = append(handlers, entry.handler)
 	}
 	return handlers
 }
diff --git a/internal/pkg/eventbus/bus_test.go b/internal/pkg/eventbus/bus_test.go
index 6590f00..4dca282 100644
--- a/internal/pkg/eventbus/bus_test.go
+++ b/internal/pkg/eventbus/bus_test.go
@@ -3,6 +3,7 @@ package eventbus
 import (
 	"context"
 	"errors"
+	"reflect"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -53,6 +54,60 @@ func TestPublishReturnsJoinedHandlerErrors(t *testing.T) {
 	}
 }
 
+func TestPublishCallsHandlersInSubscribeOrder(t *testing.T) {
+	bus := New[testEvent]()
+	calls := make([]int, 0, 3)
+
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		calls = append(calls, 1)
+		return nil
+	})
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		calls = append(calls, 2)
+		return nil
+	})
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		calls = append(calls, 3)
+		return nil
+	})
+
+	if err := bus.Publish(context.Background(), testEvent{}); err != nil {
+		t.Fatalf("publish failed: %v", err)
+	}
+
+	if !reflect.DeepEqual(calls, []int{1, 2, 3}) {
+		t.Fatalf("expected handlers to run in subscribe order, got %#v", calls)
+	}
+}
+
+func TestUnsubscribeKeepsRemainingHandlerOrder(t *testing.T) {
+	bus := New[testEvent]()
+	calls := make([]int, 0, 2)
+
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		calls = append(calls, 1)
+		return nil
+	})
+	_, unsubscribe := bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		calls = append(calls, 2)
+		return nil
+	})
+	bus.Subscribe(func(ctx context.Context, event testEvent) error {
+		calls = append(calls, 3)
+		return nil
+	})
+
+	unsubscribe()
+
+	if err := bus.Publish(context.Background(), testEvent{}); err != nil {
+		t.Fatalf("publish failed: %v", err)
+	}
+
+	if !reflect.DeepEqual(calls, []int{1, 3}) {
+		t.Fatalf("expected remaining handlers to keep order, got %#v", calls)
+	}
+}
+
 func TestPanicReturnsError(t *testing.T) {
 	bus := New[testEvent]()