fix: address comments and add a test

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Author: ferhatelmas

internal/base/queue/queue.go

@@ -26,6 +26,17 @@ import (
 	"github.com/segmentfault/pacman/log"
 )
 
+type Service[T any] interface {
+	// Send enqueues a message to be processed asynchronously.
+	Send(ctx context.Context, msg T)
+
+	// RegisterHandler sets the handler function for processing messages.
+	RegisterHandler(handler func(ctx context.Context, msg T) error)
+
+	// Close gracefully shuts down the queue, waiting for pending messages to be processed.
+	Close()
+}
+
 // Queue is a generic message queue service that processes messages asynchronously.
 // It is thread-safe and supports graceful shutdown.
 type Queue[T any] struct {
@@ -51,10 +62,9 @@ func New[T any](name string, bufferSize int) *Queue[T] {
 // It will block if the queue is full.
 func (q *Queue[T]) Send(ctx context.Context, msg T) {
 	q.mu.RLock()
-	closed := q.closed
-	q.mu.RUnlock()
+	defer q.mu.RUnlock()
 
-	if closed {
+	if q.closed {
 		log.Warnf("[%s] queue is closed, dropping message", q.name)
 		return
 	}

internal/base/queue/queue_test.go

@@ -21,6 +21,7 @@ package queue
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -209,3 +210,44 @@ func TestQueue_ConcurrentRegisterHandler(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+// TestQueue_SendCloseRace is a regression test for the race condition between
+// Send and Close. Without proper synchronization, concurrent Send and Close
+// calls could cause a "send on closed channel" panic.
+// Run with: go test -race -run TestQueue_SendCloseRace
+func TestQueue_SendCloseRace(t *testing.T) {
+	for i := range 100 {
+		t.Run(fmt.Sprintf("iteration_%d", i), func(t *testing.T) {
+			// Use large buffer to avoid blocking on channel send while holding RLock
+			q := New[*testMessage]("test-race", 1000)
+			q.RegisterHandler(func(ctx context.Context, msg *testMessage) error {
+				return nil
+			})
+
+			var wg sync.WaitGroup
+
+			// Use cancellable context so senders can exit when Close is called
+			ctx, cancel := context.WithCancel(context.Background())
+
+			// Start multiple senders
+			for j := range 10 {
+				wg.Add(1)
+				go func(id int) {
+					defer wg.Done()
+					for k := range 100 {
+						q.Send(ctx, &testMessage{ID: id*1000 + k})
+					}
+				}(j)
+			}
+
+			// Close while senders are still running
+			go func() {
+				time.Sleep(time.Microsecond * 10)
+				cancel() // Cancel context to unblock any waiting senders
+				q.Close()
+			}()
+
+			wg.Wait()
+		})
+	}
+}

internal/service/activityqueue/activity_queue.go

@@ -24,7 +24,7 @@ import (
 	"github.com/apache/answer/internal/schema"
 )
 
-type Service = *queue.Queue[*schema.ActivityMsg]
+type Service queue.Service[*schema.ActivityMsg]
 
 func NewService() Service {
 	return queue.New[*schema.ActivityMsg]("activity", 128)

internal/service/eventqueue/event_queue.go

@@ -24,7 +24,7 @@ import (
 	"github.com/apache/answer/internal/schema"
 )
 
-type Service = *queue.Queue[*schema.EventMsg]
+type Service queue.Service[*schema.EventMsg]
 
 func NewService() Service {
 	return queue.New[*schema.EventMsg]("event", 128)

internal/service/noticequeue/notice_queue.go

@@ -24,13 +24,13 @@ import (
 	"github.com/apache/answer/internal/schema"
 )
 
-type Service = *queue.Queue[*schema.NotificationMsg]
+type Service queue.Service[*schema.NotificationMsg]
 
 func NewService() Service {
 	return queue.New[*schema.NotificationMsg]("notification", 128)
 }
 
-type ExternalService = *queue.Queue[*schema.ExternalNotificationMsg]
+type ExternalService queue.Service[*schema.ExternalNotificationMsg]
 
 func NewExternalService() ExternalService {
 	return queue.New[*schema.ExternalNotificationMsg]("external_notification", 128)