Merge pull request #1169 from trheyi/main

Enhance messenger service with message handler registration and webho…

messenger/messenger.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 	"regexp"
 	"strings"
 	"sync"
@@ -35,6 +36,7 @@ type Service struct {
 	channels        map[string]types.Channel
 	defaults        map[string]string
 	receivers       map[string]context.CancelFunc // Active mail receivers by provider name
+	messageHandlers []types.MessageHandler        // Registered message handlers for OnReceive
 	mutex           sync.RWMutex
 }
 
@@ -106,6 +108,7 @@ func Load(cfg config.Config) error {
 		channels:        make(map[string]types.Channel),
 		defaults:        config.Defaults,
 		receivers:       make(map[string]context.CancelFunc),
+		messageHandlers: make([]types.MessageHandler, 0),
 	}
 
 	// Set global instance
@@ -605,8 +608,12 @@ func (m *Service) startMailReceivers() {
 					err := mp.StartMailReceiver(ctx, func(msg *types.Message) error {
 						log.Info("[Messenger] Received email via %s: Subject=%s, From=%s", providerName, msg.Subject, msg.From)
 
-						// Here you can add custom message processing logic
-						// For now, just log the received message
+						// Trigger OnReceive handlers for the received message
+						if err := m.triggerOnReceiveHandlers(ctx, msg); err != nil {
+							log.Error("[Messenger] Failed to trigger OnReceive handlers: %v", err)
+							return err
+						}
+
 						return nil
 					})
 
@@ -668,3 +675,168 @@ func (m *Service) GetActiveReceivers() []string {
 	}
 	return receivers
 }
+
+// OnReceive registers a message handler for received messages
+// Multiple handlers can be registered and will be called in order
+func (m *Service) OnReceive(handler types.MessageHandler) error {
+	if handler == nil {
+		return fmt.Errorf("handler cannot be nil")
+	}
+
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	m.messageHandlers = append(m.messageHandlers, handler)
+	log.Info("[Messenger] Registered new message handler (total: %d)", len(m.messageHandlers))
+	return nil
+}
+
+// RemoveReceiveHandler removes a previously registered message handler
+func (m *Service) RemoveReceiveHandler(handler types.MessageHandler) error {
+	if handler == nil {
+		return fmt.Errorf("handler cannot be nil")
+	}
+
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	// Find and remove the handler by comparing function pointers
+	handlerPtr := reflect.ValueOf(handler).Pointer()
+	for i, existingHandler := range m.messageHandlers {
+		if reflect.ValueOf(existingHandler).Pointer() == handlerPtr {
+			// Remove handler at index i
+			m.messageHandlers = append(m.messageHandlers[:i], m.messageHandlers[i+1:]...)
+			log.Info("[Messenger] Removed message handler (remaining: %d)", len(m.messageHandlers))
+			return nil
+		}
+	}
+
+	return fmt.Errorf("handler not found")
+}
+
+// TriggerWebhook processes incoming webhook data and triggers OnReceive handlers
+// This is used by OPENAPI endpoints to handle incoming messages
+func (m *Service) TriggerWebhook(ctx context.Context, providerName string, data map[string]interface{}) error {
+	// Get the provider to process the webhook data
+	provider, exists := m.providers[providerName]
+	if !exists {
+		return fmt.Errorf("provider not found: %s", providerName)
+	}
+
+	// First, let the provider process the webhook data
+	// This may convert webhook data into a standardized message format
+	err := provider.Receive(ctx, data)
+	if err != nil {
+		log.Warn("[Messenger] Provider %s failed to process webhook data: %v", providerName, err)
+		// Continue to trigger handlers even if provider processing fails
+	}
+
+	// Try to convert webhook data to a Message for OnReceive handlers
+	message, err := m.convertWebhookToMessage(providerName, data)
+	if err != nil {
+		log.Warn("[Messenger] Failed to convert webhook data to message: %v", err)
+		return err
+	}
+
+	// Trigger all registered OnReceive handlers
+	return m.triggerOnReceiveHandlers(ctx, message)
+}
+
+// convertWebhookToMessage attempts to convert webhook data to a standardized Message
+func (m *Service) convertWebhookToMessage(providerName string, data map[string]interface{}) (*types.Message, error) {
+	message := &types.Message{
+		Metadata: make(map[string]interface{}),
+	}
+
+	// Add provider information
+	message.Metadata["provider"] = providerName
+	message.Metadata["webhook_data"] = data
+
+	// Try to extract common fields from webhook data
+	if subject, ok := data["subject"].(string); ok {
+		message.Subject = subject
+	}
+	if from, ok := data["from"].(string); ok {
+		message.From = from
+	}
+	if body, ok := data["body"].(string); ok {
+		message.Body = body
+	}
+	if html, ok := data["html"].(string); ok {
+		message.HTML = html
+	}
+
+	// Handle "to" field which might be string or array
+	if to, ok := data["to"]; ok {
+		switch v := to.(type) {
+		case string:
+			message.To = []string{v}
+		case []string:
+			message.To = v
+		case []interface{}:
+			for _, item := range v {
+				if str, ok := item.(string); ok {
+					message.To = append(message.To, str)
+				}
+			}
+		}
+	}
+
+	// Determine message type based on provider or data
+	if msgType, ok := data["type"].(string); ok {
+		message.Type = types.MessageType(strings.ToLower(msgType))
+	} else {
+		// Default based on provider type
+		provider, exists := m.providers[providerName]
+		if exists {
+			switch strings.ToLower(provider.GetType()) {
+			case "mailer", "smtp", "mailgun":
+				message.Type = types.MessageTypeEmail
+			case "twilio":
+				// Could be SMS, WhatsApp, or Email - try to determine from data
+				if phone, ok := data["phone"].(string); ok && phone != "" {
+					message.Type = types.MessageTypeSMS
+				} else if whatsapp, ok := data["whatsapp"].(string); ok && whatsapp != "" {
+					message.Type = types.MessageTypeWhatsApp
+				} else {
+					message.Type = types.MessageTypeEmail
+				}
+			default:
+				message.Type = types.MessageTypeEmail // Default fallback
+			}
+		}
+	}
+
+	return message, nil
+}
+
+// triggerOnReceiveHandlers calls all registered OnReceive handlers
+func (m *Service) triggerOnReceiveHandlers(ctx context.Context, message *types.Message) error {
+	m.mutex.RLock()
+	handlers := make([]types.MessageHandler, len(m.messageHandlers))
+	copy(handlers, m.messageHandlers)
+	m.mutex.RUnlock()
+
+	if len(handlers) == 0 {
+		log.Debug("[Messenger] No OnReceive handlers registered")
+		return nil
+	}
+
+	log.Info("[Messenger] Triggering %d OnReceive handlers for message: %s", len(handlers), message.Subject)
+
+	var errors []string
+	for i, handler := range handlers {
+		err := handler(ctx, message)
+		if err != nil {
+			errMsg := fmt.Sprintf("handler %d failed: %v", i, err)
+			errors = append(errors, errMsg)
+			log.Error("[Messenger] %s", errMsg)
+		}
+	}
+
+	if len(errors) > 0 {
+		return fmt.Errorf("some OnReceive handlers failed: %s", strings.Join(errors, "; "))
+	}
+
+	return nil
+}

messenger/messenger_onreceive_test.go

@@ -0,0 +1,469 @@
+package messenger
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/yaoapp/yao/config"
+	"github.com/yaoapp/yao/messenger/types"
+	"github.com/yaoapp/yao/test"
+)
+
+// Test OnReceive functionality
+func TestService_OnReceive(t *testing.T) {
+	// Prepare test environment
+	test.Prepare(t, config.Conf, "YAO_TEST_APPLICATION")
+	defer test.Clean()
+
+	// Create a test service
+	service := &Service{
+		config:          &types.Config{},
+		providers:       make(map[string]types.Provider),
+		providersByType: make(map[types.MessageType][]types.Provider),
+		channels:        make(map[string]types.Channel),
+		defaults:        make(map[string]string),
+		receivers:       make(map[string]context.CancelFunc),
+		messageHandlers: make([]types.MessageHandler, 0),
+	}
+
+	// Test registering nil handler should fail
+	err := service.OnReceive(nil)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "handler cannot be nil")
+
+	// Test registering valid handlers
+	var receivedMessages []*types.Message
+	var mu sync.Mutex
+
+	handler1 := func(ctx context.Context, message *types.Message) error {
+		mu.Lock()
+		defer mu.Unlock()
+		receivedMessages = append(receivedMessages, message)
+		t.Logf("Handler 1: Received message from %s with subject: %s", message.From, message.Subject)
+		return nil
+	}
+
+	handler2 := func(ctx context.Context, message *types.Message) error {
+		mu.Lock()
+		defer mu.Unlock()
+		t.Logf("Handler 2: Processing message for analytics")
+		return nil
+	}
+
+	// Register handlers
+	err = service.OnReceive(handler1)
+	assert.NoError(t, err)
+
+	err = service.OnReceive(handler2)
+	assert.NoError(t, err)
+
+	// Verify handlers are registered
+	assert.Len(t, service.messageHandlers, 2)
+
+	// Test triggering handlers
+	testMessage := &types.Message{
+		Type:    types.MessageTypeEmail,
+		From:    "test@example.com",
+		To:      []string{"recipient@example.com"},
+		Subject: "Test Subject",
+		Body:    "Test message body",
+	}
+
+	ctx := context.Background()
+	err = service.triggerOnReceiveHandlers(ctx, testMessage)
+	assert.NoError(t, err)
+
+	// Verify message was received by handler1
+	mu.Lock()
+	assert.Len(t, receivedMessages, 1)
+	assert.Equal(t, testMessage.Subject, receivedMessages[0].Subject)
+	assert.Equal(t, testMessage.From, receivedMessages[0].From)
+	mu.Unlock()
+}
+
+func TestService_RemoveReceiveHandler(t *testing.T) {
+	// Create a test service
+	service := &Service{
+		config:          &types.Config{},
+		providers:       make(map[string]types.Provider),
+		providersByType: make(map[types.MessageType][]types.Provider),
+		channels:        make(map[string]types.Channel),
+		defaults:        make(map[string]string),
+		receivers:       make(map[string]context.CancelFunc),
+		messageHandlers: make([]types.MessageHandler, 0),
+	}
+
+	// Test removing nil handler should fail
+	err := service.RemoveReceiveHandler(nil)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "handler cannot be nil")
+
+	// Create and register a handler
+	handler := func(ctx context.Context, message *types.Message) error {
+		return nil
+	}
+
+	err = service.OnReceive(handler)
+	assert.NoError(t, err)
+	assert.Len(t, service.messageHandlers, 1)
+
+	// Remove the handler
+	err = service.RemoveReceiveHandler(handler)
+	assert.NoError(t, err)
+	assert.Len(t, service.messageHandlers, 0)
+
+	// Try to remove the same handler again should fail
+	err = service.RemoveReceiveHandler(handler)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "handler not found")
+}
+
+func TestService_TriggerWebhook(t *testing.T) {
+	// Prepare test environment
+	test.Prepare(t, config.Conf, "YAO_TEST_APPLICATION")
+	defer test.Clean()
+
+	// Load real providers
+	providers, err := loadProviders()
+	require.NoError(t, err)
+
+	// Create a test service with real providers
+	service := &Service{
+		config:          &types.Config{},
+		providers:       providers,
+		providersByType: make(map[types.MessageType][]types.Provider),
+		channels:        make(map[string]types.Channel),
+		defaults:        make(map[string]string),
+		receivers:       make(map[string]context.CancelFunc),
+		messageHandlers: make([]types.MessageHandler, 0),
+	}
+
+	// Test with non-existent provider
+	ctx := context.Background()
+	webhookData := map[string]interface{}{
+		"from":    "test@example.com",
+		"to":      "recipient@example.com",
+		"subject": "Test Subject",
+		"body":    "Test message body",
+	}
+
+	err = service.TriggerWebhook(ctx, "nonexistent", webhookData)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "provider not found")
+
+	// Test with existing provider (if any are loaded)
+	if len(providers) > 0 {
+		// Get the first provider name
+		var providerName string
+		for name := range providers {
+			providerName = name
+			break
+		}
+
+		// Register a handler to capture the triggered message
+		var receivedMessage *types.Message
+		var mu sync.Mutex
+
+		handler := func(ctx context.Context, message *types.Message) error {
+			mu.Lock()
+			defer mu.Unlock()
+			receivedMessage = message
+			t.Logf("Webhook handler: Received message from %s with subject: %s", message.From, message.Subject)
+			return nil
+		}
+
+		err = service.OnReceive(handler)
+		assert.NoError(t, err)
+
+		// Trigger webhook
+		err = service.TriggerWebhook(ctx, providerName, webhookData)
+		// Note: This might fail if the provider's Receive method has validation,
+		// but it should not panic and should attempt to trigger handlers
+		if err != nil {
+			t.Logf("TriggerWebhook returned error (may be expected): %v", err)
+		}
+
+		// Give some time for async processing
+		time.Sleep(100 * time.Millisecond)
+
+		// Check if handler was triggered
+		mu.Lock()
+		if receivedMessage != nil {
+			assert.Equal(t, "test@example.com", receivedMessage.From)
+			assert.Equal(t, "Test Subject", receivedMessage.Subject)
+			assert.Equal(t, "Test message body", receivedMessage.Body)
+			assert.Contains(t, receivedMessage.To, "recipient@example.com")
+		}
+		mu.Unlock()
+	}
+}
+
+func TestService_ConvertWebhookToMessage(t *testing.T) {
+	// Prepare test environment
+	test.Prepare(t, config.Conf, "YAO_TEST_APPLICATION")
+	defer test.Clean()
+
+	// Load real providers
+	providers, err := loadProviders()
+	require.NoError(t, err)
+
+	// Create a test service
+	service := &Service{
+		config:          &types.Config{},
+		providers:       providers,
+		providersByType: make(map[types.MessageType][]types.Provider),
+		channels:        make(map[string]types.Channel),
+		defaults:        make(map[string]string),
+		receivers:       make(map[string]context.CancelFunc),
+		messageHandlers: make([]types.MessageHandler, 0),
+	}
+
+	tests := []struct {
+		name         string
+		providerName string
+		data         map[string]interface{}
+		expectedType types.MessageType
+	}{
+		{
+			name:         "Email webhook data",
+			providerName: "test-mailer",
+			data: map[string]interface{}{
+				"type":    "email",
+				"from":    "sender@example.com",
+				"to":      "recipient@example.com",
+				"subject": "Test Email",
+				"body":    "Email body content",
+				"html":    "<p>Email HTML content</p>",
+			},
+			expectedType: types.MessageTypeEmail,
+		},
+		{
+			name:         "SMS webhook data",
+			providerName: "test-twilio",
+			data: map[string]interface{}{
+				"type":  "sms",
+				"from":  "+1234567890",
+				"to":    "+0987654321",
+				"body":  "SMS message content",
+				"phone": "+0987654321",
+			},
+			expectedType: types.MessageTypeSMS,
+		},
+		{
+			name:         "WhatsApp webhook data",
+			providerName: "test-twilio",
+			data: map[string]interface{}{
+				"type":     "whatsapp",
+				"from":     "+1234567890",
+				"to":       "+0987654321",
+				"body":     "WhatsApp message content",
+				"whatsapp": "+0987654321",
+			},
+			expectedType: types.MessageTypeWhatsApp,
+		},
+		{
+			name:         "Array recipients",
+			providerName: "test-mailer",
+			data: map[string]interface{}{
+				"from":    "sender@example.com",
+				"to":      []string{"recipient1@example.com", "recipient2@example.com"},
+				"subject": "Test Email",
+				"body":    "Email body content",
+			},
+			expectedType: types.MessageTypeEmail,
+		},
+		{
+			name:         "Interface array recipients",
+			providerName: "test-mailer",
+			data: map[string]interface{}{
+				"from":    "sender@example.com",
+				"to":      []interface{}{"recipient1@example.com", "recipient2@example.com"},
+				"subject": "Test Email",
+				"body":    "Email body content",
+			},
+			expectedType: types.MessageTypeEmail,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			message, err := service.convertWebhookToMessage(tt.providerName, tt.data)
+			assert.NoError(t, err)
+			assert.NotNil(t, message)
+
+			// Check basic fields
+			if from, ok := tt.data["from"].(string); ok {
+				assert.Equal(t, from, message.From)
+			}
+			if subject, ok := tt.data["subject"].(string); ok {
+				assert.Equal(t, subject, message.Subject)
+			}
+			if body, ok := tt.data["body"].(string); ok {
+				assert.Equal(t, body, message.Body)
+			}
+			if html, ok := tt.data["html"].(string); ok {
+				assert.Equal(t, html, message.HTML)
+			}
+
+			// Check recipients
+			if to, ok := tt.data["to"]; ok {
+				switch v := to.(type) {
+				case string:
+					assert.Contains(t, message.To, v)
+				case []string:
+					for _, recipient := range v {
+						assert.Contains(t, message.To, recipient)
+					}
+				case []interface{}:
+					for _, recipient := range v {
+						if str, ok := recipient.(string); ok {
+							assert.Contains(t, message.To, str)
+						}
+					}
+				}
+			}
+
+			// Check message type
+			if tt.data["type"] != nil {
+				assert.Equal(t, tt.expectedType, message.Type)
+			}
+
+			// Check metadata
+			assert.NotNil(t, message.Metadata)
+			assert.Equal(t, tt.providerName, message.Metadata["provider"])
+			assert.Equal(t, tt.data, message.Metadata["webhook_data"])
+		})
+	}
+}
+
+func TestService_TriggerOnReceiveHandlers_ErrorHandling(t *testing.T) {
+	// Create a test service
+	service := &Service{
+		config:          &types.Config{},
+		providers:       make(map[string]types.Provider),
+		providersByType: make(map[types.MessageType][]types.Provider),
+		channels:        make(map[string]types.Channel),
+		defaults:        make(map[string]string),
+		receivers:       make(map[string]context.CancelFunc),
+		messageHandlers: make([]types.MessageHandler, 0),
+	}
+
+	// Test with no handlers
+	ctx := context.Background()
+	testMessage := &types.Message{
+		Type:    types.MessageTypeEmail,
+		From:    "test@example.com",
+		To:      []string{"recipient@example.com"},
+		Subject: "Test Subject",
+		Body:    "Test message body",
+	}
+
+	err := service.triggerOnReceiveHandlers(ctx, testMessage)
+	assert.NoError(t, err)
+
+	// Register handlers with different behaviors
+	successHandler := func(ctx context.Context, message *types.Message) error {
+		t.Logf("Success handler: %s", message.Subject)
+		return nil
+	}
+
+	errorHandler := func(ctx context.Context, message *types.Message) error {
+		t.Logf("Error handler: %s", message.Subject)
+		return assert.AnError
+	}
+
+	anotherSuccessHandler := func(ctx context.Context, message *types.Message) error {
+		t.Logf("Another success handler: %s", message.Subject)
+		return nil
+	}
+
+	// Register handlers
+	err = service.OnReceive(successHandler)
+	assert.NoError(t, err)
+
+	err = service.OnReceive(errorHandler)
+	assert.NoError(t, err)
+
+	err = service.OnReceive(anotherSuccessHandler)
+	assert.NoError(t, err)
+
+	// Trigger handlers - should continue even if one fails
+	err = service.triggerOnReceiveHandlers(ctx, testMessage)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "some OnReceive handlers failed")
+	assert.Contains(t, err.Error(), "handler 1 failed")
+}
+
+// Integration test with real messenger instance
+func TestMessenger_OnReceiveIntegration(t *testing.T) {
+	// Prepare test environment
+	test.Prepare(t, config.Conf, "YAO_TEST_APPLICATION")
+	defer test.Clean()
+
+	// Load messenger configuration
+	err := Load(config.Conf)
+	require.NoError(t, err)
+	require.NotNil(t, Instance)
+
+	// Cast to Service to access our new methods
+	service, ok := Instance.(*Service)
+	require.True(t, ok, "Instance should be of type *Service")
+
+	// Test OnReceive with real instance
+	var receivedMessage *types.Message
+	var mu sync.Mutex
+
+	handler := func(ctx context.Context, message *types.Message) error {
+		mu.Lock()
+		defer mu.Unlock()
+		receivedMessage = message
+		t.Logf("Integration handler: Received message from %s", message.From)
+		return nil
+	}
+
+	err = service.OnReceive(handler)
+	assert.NoError(t, err)
+
+	// Test TriggerWebhook with real providers (if any exist)
+	if len(service.providers) > 0 {
+		// Get the first provider name
+		var providerName string
+		for name := range service.providers {
+			providerName = name
+			break
+		}
+
+		webhookData := map[string]interface{}{
+			"from":    "integration@example.com",
+			"to":      "test@example.com",
+			"subject": "Integration Test",
+			"body":    "Integration test message",
+		}
+
+		ctx := context.Background()
+		err = service.TriggerWebhook(ctx, providerName, webhookData)
+		// Error is acceptable as provider might reject test data
+		if err != nil {
+			t.Logf("TriggerWebhook returned error (may be expected): %v", err)
+		}
+
+		// Give some time for processing
+		time.Sleep(100 * time.Millisecond)
+
+		// Check if handler was triggered
+		mu.Lock()
+		if receivedMessage != nil {
+			assert.Equal(t, "integration@example.com", receivedMessage.From)
+			assert.Equal(t, "Integration Test", receivedMessage.Subject)
+		}
+		mu.Unlock()
+	}
+
+	// Clean up - remove handler
+	err = service.RemoveReceiveHandler(handler)
+	assert.NoError(t, err)
+}

messenger/messenger_test.go

@@ -177,8 +177,8 @@ func TestProviderTypes(t *testing.T) {
 		providerName string
 		expectedType string
 	}{
-		{"primary", "smtp"},      // Generated from primary.smtp.yao
-		{"reliable", "smtp"},     // Generated from reliable.smtp.yao
+		{"primary", "mailer"},    // Generated from primary.mailer.yao
+		{"reliable", "mailer"},   // Generated from reliable.mailer.yao
 		{"marketing", "mailgun"}, // Generated from marketing.mailgun.yao
 		{"unified", "twilio"},    // Generated from unified.twilio.yao
 	}

messenger/types/interfaces.go

@@ -2,6 +2,9 @@ package types
 
 import "context"
 
+// MessageHandler defines a callback function for handling received messages
+type MessageHandler func(ctx context.Context, message *Message) error
+
 // Provider defines the interface for message providers
 type Provider interface {
 	// Send sends a message using the provider
@@ -46,6 +49,17 @@ type Messenger interface {
 	// GetChannels returns all available channels
 	GetChannels() []string
 
+	// OnReceive registers a message handler for received messages
+	// Multiple handlers can be registered and will be called in order
+	OnReceive(handler MessageHandler) error
+
+	// RemoveReceiveHandler removes a previously registered message handler
+	RemoveReceiveHandler(handler MessageHandler) error
+
+	// TriggerWebhook processes incoming webhook data and triggers OnReceive handlers
+	// This is used by OPENAPI endpoints to handle incoming messages
+	TriggerWebhook(ctx context.Context, providerName string, data map[string]interface{}) error
+
 	// Close closes all provider connections
 	Close() error
 }