From ffc666c4de7a021801159b4b11fa5c3c5749bcfa Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 28 Sep 2025 10:00:38 +0800 Subject: [PATCH] Enhance messenger service with message handler registration and webhook processing - Introduced OnReceive method to register multiple message handlers for processing received messages. - Added RemoveReceiveHandler method to unregister specific message handlers. - Implemented TriggerWebhook method to process incoming webhook data and trigger registered handlers. - Updated existing tests to reflect changes in provider types from 'smtp' to 'mailer' for consistency. --- messenger/messenger.go | 176 +++++++++- messenger/messenger_onreceive_test.go | 469 ++++++++++++++++++++++++++ messenger/messenger_test.go | 4 +- messenger/types/interfaces.go | 14 + 4 files changed, 659 insertions(+), 4 deletions(-) create mode 100644 messenger/messenger_onreceive_test.go diff --git a/messenger/messenger.go b/messenger/messenger.go index c7877e61..dfe6bd77 100644 --- a/messenger/messenger.go +++ b/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 +} diff --git a/messenger/messenger_onreceive_test.go b/messenger/messenger_onreceive_test.go new file mode 100644 index 00000000..ac7bcb67 --- /dev/null +++ b/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": "

Email HTML content

", + }, + 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) +} diff --git a/messenger/messenger_test.go b/messenger/messenger_test.go index 7e0f3344..a632c22b 100644 --- a/messenger/messenger_test.go +++ b/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 } diff --git a/messenger/types/interfaces.go b/messenger/types/interfaces.go index 6592b609..ff8733a3 100644 --- a/messenger/types/interfaces.go +++ b/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 }