Merge pull request #1169 from trheyi/main

Enhance messenger service with message handler registration and webho…

Author: trheyi

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
+}