chore: resolve conflicts with upstream/main

2026-06-12 18:08:54 +00:00 · 2026-02-28 12:21:54 +08:00
parent f2a71ca824 2f4f45080b
commit 62bdece7f5
96 changed files with 11203 additions and 1610 deletions
@@ -2,81 +2,156 @@ package bus

 import (
 	"context"
-	"sync"
+	"errors"
+	"sync/atomic"
+
+	"github.com/sipeed/picoclaw/pkg/logger"
 )

+// ErrBusClosed is returned when publishing to a closed MessageBus.
+var ErrBusClosed = errors.New("message bus closed")
+
+const defaultBusBufferSize = 64
+
 type MessageBus struct {
-	inbound  chan InboundMessage
-	outbound chan OutboundMessage
-	handlers map[string]MessageHandler
-	closed   bool
-	mu       sync.RWMutex
+	inbound       chan InboundMessage
+	outbound      chan OutboundMessage
+	outboundMedia chan OutboundMediaMessage
+	done          chan struct{}
+	closed        atomic.Bool
 }

 func NewMessageBus() *MessageBus {
 	return &MessageBus{
-		inbound:  make(chan InboundMessage, 100),
-		outbound: make(chan OutboundMessage, 100),
-		handlers: make(map[string]MessageHandler),
+		inbound:       make(chan InboundMessage, defaultBusBufferSize),
+		outbound:      make(chan OutboundMessage, defaultBusBufferSize),
+		outboundMedia: make(chan OutboundMediaMessage, defaultBusBufferSize),
+		done:          make(chan struct{}),
 	}
 }

-func (mb *MessageBus) PublishInbound(msg InboundMessage) {
-	mb.mu.RLock()
-	defer mb.mu.RUnlock()
-	if mb.closed {
-		return
+func (mb *MessageBus) PublishInbound(ctx context.Context, msg InboundMessage) error {
+	if mb.closed.Load() {
+		return ErrBusClosed
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	select {
+	case mb.inbound <- msg:
+		return nil
+	case <-mb.done:
+		return ErrBusClosed
+	case <-ctx.Done():
+		return ctx.Err()
 	}
-	mb.inbound <- msg
 }

 func (mb *MessageBus) ConsumeInbound(ctx context.Context) (InboundMessage, bool) {
 	select {
-	case msg := <-mb.inbound:
-		return msg, true
+	case msg, ok := <-mb.inbound:
+		return msg, ok
+	case <-mb.done:
+		return InboundMessage{}, false
 	case <-ctx.Done():
 		return InboundMessage{}, false
 	}
 }

-func (mb *MessageBus) PublishOutbound(msg OutboundMessage) {
-	mb.mu.RLock()
-	defer mb.mu.RUnlock()
-	if mb.closed {
-		return
+func (mb *MessageBus) PublishOutbound(ctx context.Context, msg OutboundMessage) error {
+	if mb.closed.Load() {
+		return ErrBusClosed
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	select {
+	case mb.outbound <- msg:
+		return nil
+	case <-mb.done:
+		return ErrBusClosed
+	case <-ctx.Done():
+		return ctx.Err()
 	}
-	mb.outbound <- msg
 }

 func (mb *MessageBus) SubscribeOutbound(ctx context.Context) (OutboundMessage, bool) {
 	select {
-	case msg := <-mb.outbound:
-		return msg, true
+	case msg, ok := <-mb.outbound:
+		return msg, ok
+	case <-mb.done:
+		return OutboundMessage{}, false
 	case <-ctx.Done():
 		return OutboundMessage{}, false
 	}
 }

-func (mb *MessageBus) RegisterHandler(channel string, handler MessageHandler) {
-	mb.mu.Lock()
-	defer mb.mu.Unlock()
-	mb.handlers[channel] = handler
+func (mb *MessageBus) PublishOutboundMedia(ctx context.Context, msg OutboundMediaMessage) error {
+	if mb.closed.Load() {
+		return ErrBusClosed
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	select {
+	case mb.outboundMedia <- msg:
+		return nil
+	case <-mb.done:
+		return ErrBusClosed
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }

-func (mb *MessageBus) GetHandler(channel string) (MessageHandler, bool) {
-	mb.mu.RLock()
-	defer mb.mu.RUnlock()
-	handler, ok := mb.handlers[channel]
-	return handler, ok
+func (mb *MessageBus) SubscribeOutboundMedia(ctx context.Context) (OutboundMediaMessage, bool) {
+	select {
+	case msg, ok := <-mb.outboundMedia:
+		return msg, ok
+	case <-mb.done:
+		return OutboundMediaMessage{}, false
+	case <-ctx.Done():
+		return OutboundMediaMessage{}, false
+	}
 }

 func (mb *MessageBus) Close() {
-	mb.mu.Lock()
-	defer mb.mu.Unlock()
-	if mb.closed {
-		return
+	if mb.closed.CompareAndSwap(false, true) {
+		close(mb.done)
+
+		// Drain buffered channels so messages aren't silently lost.
+		// Channels are NOT closed to avoid send-on-closed panics from concurrent publishers.
+		drained := 0
+		for {
+			select {
+			case <-mb.inbound:
+				drained++
+			default:
+				goto doneInbound
+			}
+		}
+	doneInbound:
+		for {
+			select {
+			case <-mb.outbound:
+				drained++
+			default:
+				goto doneOutbound
+			}
+		}
+	doneOutbound:
+		for {
+			select {
+			case <-mb.outboundMedia:
+				drained++
+			default:
+				goto doneMedia
+			}
+		}
+	doneMedia:
+		if drained > 0 {
+			logger.DebugCF("bus", "Drained buffered messages during close", map[string]any{
+				"count": drained,
+			})
+		}
 	}
-	mb.closed = true
-	close(mb.inbound)
-	close(mb.outbound)
 }
@@ -0,0 +1,229 @@
+package bus
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestPublishConsume(t *testing.T) {
+	mb := NewMessageBus()
+	defer mb.Close()
+
+	ctx := context.Background()
+
+	msg := InboundMessage{
+		Channel:  "test",
+		SenderID: "user1",
+		ChatID:   "chat1",
+		Content:  "hello",
+	}
+
+	if err := mb.PublishInbound(ctx, msg); err != nil {
+		t.Fatalf("PublishInbound failed: %v", err)
+	}
+
+	got, ok := mb.ConsumeInbound(ctx)
+	if !ok {
+		t.Fatal("ConsumeInbound returned ok=false")
+	}
+	if got.Content != "hello" {
+		t.Fatalf("expected content 'hello', got %q", got.Content)
+	}
+	if got.Channel != "test" {
+		t.Fatalf("expected channel 'test', got %q", got.Channel)
+	}
+}
+
+func TestPublishOutboundSubscribe(t *testing.T) {
+	mb := NewMessageBus()
+	defer mb.Close()
+
+	ctx := context.Background()
+
+	msg := OutboundMessage{
+		Channel: "telegram",
+		ChatID:  "123",
+		Content: "world",
+	}
+
+	if err := mb.PublishOutbound(ctx, msg); err != nil {
+		t.Fatalf("PublishOutbound failed: %v", err)
+	}
+
+	got, ok := mb.SubscribeOutbound(ctx)
+	if !ok {
+		t.Fatal("SubscribeOutbound returned ok=false")
+	}
+	if got.Content != "world" {
+		t.Fatalf("expected content 'world', got %q", got.Content)
+	}
+}
+
+func TestPublishInbound_ContextCancel(t *testing.T) {
+	mb := NewMessageBus()
+	defer mb.Close()
+
+	// Fill the buffer
+	ctx := context.Background()
+	for i := range defaultBusBufferSize {
+		if err := mb.PublishInbound(ctx, InboundMessage{Content: "fill"}); err != nil {
+			t.Fatalf("fill failed at %d: %v", i, err)
+		}
+	}
+
+	// Now buffer is full; publish with a canceled context
+	cancelCtx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := mb.PublishInbound(cancelCtx, InboundMessage{Content: "overflow"})
+	if err == nil {
+		t.Fatal("expected error from canceled context, got nil")
+	}
+	if err != context.Canceled {
+		t.Fatalf("expected context.Canceled, got %v", err)
+	}
+}
+
+func TestPublishInbound_BusClosed(t *testing.T) {
+	mb := NewMessageBus()
+	mb.Close()
+
+	err := mb.PublishInbound(context.Background(), InboundMessage{Content: "test"})
+	if err != ErrBusClosed {
+		t.Fatalf("expected ErrBusClosed, got %v", err)
+	}
+}
+
+func TestPublishOutbound_BusClosed(t *testing.T) {
+	mb := NewMessageBus()
+	mb.Close()
+
+	err := mb.PublishOutbound(context.Background(), OutboundMessage{Content: "test"})
+	if err != ErrBusClosed {
+		t.Fatalf("expected ErrBusClosed, got %v", err)
+	}
+}
+
+func TestConsumeInbound_ContextCancel(t *testing.T) {
+	mb := NewMessageBus()
+	defer mb.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, ok := mb.ConsumeInbound(ctx)
+	if ok {
+		t.Fatal("expected ok=false when context is canceled")
+	}
+}
+
+func TestConsumeInbound_BusClosed(t *testing.T) {
+	mb := NewMessageBus()
+	mb.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	_, ok := mb.ConsumeInbound(ctx)
+	if ok {
+		t.Fatal("expected ok=false when bus is closed")
+	}
+}
+
+func TestSubscribeOutbound_BusClosed(t *testing.T) {
+	mb := NewMessageBus()
+	mb.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	_, ok := mb.SubscribeOutbound(ctx)
+	if ok {
+		t.Fatal("expected ok=false when bus is closed")
+	}
+}
+
+func TestConcurrentPublishClose(t *testing.T) {
+	mb := NewMessageBus()
+	ctx := context.Background()
+
+	const numGoroutines = 100
+	var wg sync.WaitGroup
+	wg.Add(numGoroutines + 1)
+
+	// Spawn many goroutines trying to publish
+	for range numGoroutines {
+		go func() {
+			defer wg.Done()
+			// Use a short timeout context so we don't block forever after close
+			publishCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
+			defer cancel()
+			// Errors are expected; we just must not panic or deadlock
+			_ = mb.PublishInbound(publishCtx, InboundMessage{Content: "concurrent"})
+		}()
+	}
+
+	// Close from another goroutine
+	go func() {
+		defer wg.Done()
+		time.Sleep(5 * time.Millisecond)
+		mb.Close()
+	}()
+
+	// Must complete without deadlock
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// success
+	case <-time.After(5 * time.Second):
+		t.Fatal("test timed out - possible deadlock")
+	}
+}
+
+func TestPublishInbound_FullBuffer(t *testing.T) {
+	mb := NewMessageBus()
+	defer mb.Close()
+
+	ctx := context.Background()
+
+	// Fill the buffer
+	for i := range defaultBusBufferSize {
+		if err := mb.PublishInbound(ctx, InboundMessage{Content: "fill"}); err != nil {
+			t.Fatalf("fill failed at %d: %v", i, err)
+		}
+	}
+
+	// Buffer is full; publish with short timeout
+	timeoutCtx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+
+	err := mb.PublishInbound(timeoutCtx, InboundMessage{Content: "overflow"})
+	if err == nil {
+		t.Fatal("expected error when buffer is full and context times out")
+	}
+	if err != context.DeadlineExceeded {
+		t.Fatalf("expected context.DeadlineExceeded, got %v", err)
+	}
+}
+
+func TestCloseIdempotent(t *testing.T) {
+	mb := NewMessageBus()
+
+	// Multiple Close calls must not panic
+	mb.Close()
+	mb.Close()
+	mb.Close()
+
+	// After close, publish should return ErrBusClosed
+	err := mb.PublishInbound(context.Background(), InboundMessage{Content: "test"})
+	if err != ErrBusClosed {
+		t.Fatalf("expected ErrBusClosed after multiple closes, got %v", err)
+	}
+}
@@ -1,11 +1,30 @@
 package bus

+// Peer identifies the routing peer for a message (direct, group, channel, etc.)
+type Peer struct {
+	Kind string `json:"kind"` // "direct" | "group" | "channel" | ""
+	ID   string `json:"id"`
+}
+
+// SenderInfo provides structured sender identity information.
+type SenderInfo struct {
+	Platform    string `json:"platform,omitempty"`     // "telegram", "discord", "slack", ...
+	PlatformID  string `json:"platform_id,omitempty"`  // raw platform ID, e.g. "123456"
+	CanonicalID string `json:"canonical_id,omitempty"` // "platform:id" format
+	Username    string `json:"username,omitempty"`     // username (e.g. @alice)
+	DisplayName string `json:"display_name,omitempty"` // display name
+}
+
 type InboundMessage struct {
 	Channel    string            `json:"channel"`
 	SenderID   string            `json:"sender_id"`
+	Sender     SenderInfo        `json:"sender"`
 	ChatID     string            `json:"chat_id"`
 	Content    string            `json:"content"`
 	Media      []string          `json:"media,omitempty"`
+	Peer       Peer              `json:"peer"`                  // routing peer
+	MessageID  string            `json:"message_id,omitempty"`  // platform message ID
+	MediaScope string            `json:"media_scope,omitempty"` // media lifecycle scope
 	SessionKey string            `json:"session_key"`
 	Metadata   map[string]string `json:"metadata,omitempty"`
 }
@@ -16,4 +35,18 @@ type OutboundMessage struct {
 	Content string `json:"content"`
 }

-type MessageHandler func(InboundMessage) error
+// MediaPart describes a single media attachment to send.
+type MediaPart struct {
+	Type        string `json:"type"`                   // "image" | "audio" | "video" | "file"
+	Ref         string `json:"ref"`                    // media store ref, e.g. "media://abc123"
+	Caption     string `json:"caption,omitempty"`      // optional caption text
+	Filename    string `json:"filename,omitempty"`     // original filename hint
+	ContentType string `json:"content_type,omitempty"` // MIME type hint
+}
+
+// OutboundMediaMessage carries media attachments from Agent to channels via the bus.
+type OutboundMediaMessage struct {
+	Channel string      `json:"channel"`
+	ChatID  string      `json:"chat_id"`
+	Parts   []MediaPart `json:"parts"`
+}