From 9e0afa3f424390a618c5e5e4097d496b218af28d Mon Sep 17 00:00:00 2001 From: Max Date: Thu, 25 Sep 2025 10:33:35 +0800 Subject: [PATCH] Implement team member invitation expiration handling - Added functionality to manage expiration of member invitations, ensuring that outdated invitations are automatically invalidated. - Updated invitation-related data structures to include expiration timestamps for better tracking. - Enhanced the invitation workflow by implementing checks for expired invitations during member management operations. - Refactored related tests to cover new expiration logic, ensuring robust validation of invitation handling. --- messenger/messenger.go | 490 +++++++++++++++++++++++++ messenger/providers/mailgun/mailgun.go | 197 ++++++++++ messenger/providers/smtp/smtp.go | 286 +++++++++++++++ messenger/providers/twilio/twilio.go | 408 ++++++++++++++++++++ messenger/types/interfaces.go | 46 +++ messenger/types/types.go | 119 ++++++ 6 files changed, 1546 insertions(+) create mode 100644 messenger/messenger.go create mode 100644 messenger/providers/mailgun/mailgun.go create mode 100644 messenger/providers/smtp/smtp.go create mode 100644 messenger/providers/twilio/twilio.go create mode 100644 messenger/types/interfaces.go create mode 100644 messenger/types/types.go diff --git a/messenger/messenger.go b/messenger/messenger.go new file mode 100644 index 00000000..14037624 --- /dev/null +++ b/messenger/messenger.go @@ -0,0 +1,490 @@ +package messenger + +import ( + "fmt" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/yaoapp/gou/application" + "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/config" + "github.com/yaoapp/yao/messenger/providers/mailgun" + "github.com/yaoapp/yao/messenger/providers/smtp" + "github.com/yaoapp/yao/messenger/providers/twilio" + "github.com/yaoapp/yao/messenger/types" + "github.com/yaoapp/yao/share" +) + +// Instance is the global messenger instance +var Instance types.Messenger = nil + +// Pools holds all loaded providers +var Pools = map[string]types.Provider{} +var rwlock sync.RWMutex + +// Service implements the Messenger interface +type Service struct { + config *types.Config + providers map[string]types.Provider // All providers by name + providersByType map[types.MessageType][]types.Provider // Providers grouped by message type + channels map[string]types.Channel + defaults map[string]string + mutex sync.RWMutex +} + +// Load loads the messenger configuration and providers +func Load(cfg config.Config) error { + // Check if messengers directory exists + exists, err := application.App.Exists("messengers") + if err != nil { + return err + } + if !exists { + log.Warn("[Messenger] messengers directory not found, skip loading messenger") + return nil + } + + // Load channels configuration + channelsPath := filepath.Join("messengers", "channels.yao") + var channelsConfig map[string]interface{} + if exists, _ := application.App.Exists(channelsPath); exists { + raw, err := application.App.Read(channelsPath) + if err != nil { + return err + } + err = application.Parse("channels.yao", raw, &channelsConfig) + if err != nil { + return err + } + } + + // Load provider configurations + providers, err := loadProviders() + if err != nil { + return err + } + + // Create messenger configuration + config := &types.Config{ + Providers: []types.ProviderConfig{}, + Channels: make(map[string]types.Channel), + Defaults: make(map[string]string), + Global: types.GlobalConfig{ + RetryAttempts: 3, + RetryDelay: time.Second * 2, + Timeout: time.Second * 30, + LogLevel: "info", + }, + } + + // Parse channels configuration and convert to defaults map + if channelsConfig != nil { + parseChannelsConfig(channelsConfig, config.Defaults) + } + + // Group providers by message type + providersByType := make(map[types.MessageType][]types.Provider) + for _, provider := range providers { + // Determine which message types this provider supports + supportedTypes := getSupportedMessageTypes(provider) + for _, msgType := range supportedTypes { + providersByType[msgType] = append(providersByType[msgType], provider) + } + } + + // Create messenger service + service := &Service{ + config: config, + providers: providers, + providersByType: providersByType, + channels: make(map[string]types.Channel), + defaults: config.Defaults, + } + + // Set global instance + Instance = service + return nil +} + +// loadProviders loads all provider configurations from the providers directory +func loadProviders() (map[string]types.Provider, error) { + providers := make(map[string]types.Provider) + + // Check if providers directory exists + providersPath := "messengers/providers" + exists, err := application.App.Exists(providersPath) + if err != nil { + return providers, err + } + if !exists { + return providers, nil + } + + // Walk through provider files + messages := []string{} + exts := []string{"*.yao", "*.json", "*.jsonc"} + err = application.App.Walk(providersPath, func(root, file string, isdir bool) error { + if isdir { + return nil + } + + provider, err := loadProvider(file, share.ID(root, file)) + if err != nil { + messages = append(messages, err.Error()) + return nil // Continue loading other providers + } + + if provider != nil { + providers[provider.GetName()] = provider + } + return nil + }, exts...) + + if len(messages) > 0 { + log.Warn("[Messenger] Some providers failed to load: %s", strings.Join(messages, "; ")) + } + + return providers, err +} + +// loadProvider loads a single provider configuration +func loadProvider(file string, name string) (types.Provider, error) { + raw, err := application.App.Read(file) + if err != nil { + return nil, err + } + + var config types.ProviderConfig + err = application.Parse(file, raw, &config) + if err != nil { + return nil, err + } + + // Set name if not provided + if config.Name == "" { + config.Name = name + } + + // Create provider based on type + return createProvider(config) +} + +// createProvider creates a provider instance based on configuration +func createProvider(config types.ProviderConfig) (types.Provider, error) { + // Default to enabled if not specified + if !config.Enabled && config.Enabled != false { + config.Enabled = true + } + + if !config.Enabled { + return nil, nil + } + + // Use connector field to determine provider type + connector := strings.ToLower(config.Connector) + + // Create provider based on connector + switch connector { + case "smtp": + return smtp.NewSMTPProvider(config) + case "twilio": + return createTwilioProvider(config) + case "mailgun": + return mailgun.NewMailgunProvider(config) + default: + return nil, fmt.Errorf("unsupported connector: %s", connector) + } +} + +// createTwilioProvider creates a unified Twilio provider that handles all message types +func createTwilioProvider(config types.ProviderConfig) (types.Provider, error) { + return twilio.NewTwilioProvider(config) +} + +// Send sends a message using the specified channel or default provider +func (m *Service) Send(channel string, message *types.Message) error { + m.mutex.RLock() + defer m.mutex.RUnlock() + + // Get provider for channel + providerName := m.getProviderForChannel(channel, string(message.Type)) + if providerName == "" { + return fmt.Errorf("no provider configured for channel: %s, type: %s", channel, message.Type) + } + + return m.SendWithProvider(providerName, message) +} + +// SendWithProvider sends a message using a specific provider +func (m *Service) SendWithProvider(providerName string, message *types.Message) error { + provider, exists := m.providers[providerName] + if !exists { + return fmt.Errorf("provider not found: %s", providerName) + } + + // Validate message + if err := m.validateMessage(message); err != nil { + return fmt.Errorf("message validation failed: %w", err) + } + + // Send message with retry logic + var lastErr error + maxAttempts := m.config.Global.RetryAttempts + if maxAttempts <= 0 { + maxAttempts = 1 + } + + for attempt := 1; attempt <= maxAttempts; attempt++ { + err := provider.Send(message) + if err == nil { + log.Info("[Messenger] Message sent successfully via %s (attempt %d/%d)", providerName, attempt, maxAttempts) + return nil + } + + lastErr = err + if attempt < maxAttempts { + log.Warn("[Messenger] Send attempt %d/%d failed for provider %s: %v", attempt, maxAttempts, providerName, err) + time.Sleep(m.config.Global.RetryDelay) + } + } + + return fmt.Errorf("failed to send message after %d attempts: %w", maxAttempts, lastErr) +} + +// SendBatch sends multiple messages in batch +func (m *Service) SendBatch(channel string, messages []*types.Message) error { + if len(messages) == 0 { + return nil + } + + // Group messages by provider + providerMessages := make(map[string][]*types.Message) + for _, message := range messages { + providerName := m.getProviderForChannel(channel, string(message.Type)) + if providerName == "" { + return fmt.Errorf("no provider configured for channel: %s, type: %s", channel, message.Type) + } + providerMessages[providerName] = append(providerMessages[providerName], message) + } + + // Send messages by provider + var errors []string + for providerName, msgs := range providerMessages { + provider, exists := m.providers[providerName] + if !exists { + errors = append(errors, fmt.Sprintf("provider not found: %s", providerName)) + continue + } + + err := provider.SendBatch(msgs) + if err != nil { + errors = append(errors, fmt.Sprintf("provider %s: %v", providerName, err)) + } + } + + if len(errors) > 0 { + return fmt.Errorf("batch send errors: %s", strings.Join(errors, "; ")) + } + return nil +} + +// GetProvider returns a provider by name +func (m *Service) GetProvider(name string) (types.Provider, error) { + m.mutex.RLock() + defer m.mutex.RUnlock() + + provider, exists := m.providers[name] + if !exists { + return nil, fmt.Errorf("provider not found: %s", name) + } + return provider, nil +} + +// GetProviders returns all providers for a message type +func (m *Service) GetProviders(messageType string) []types.Provider { + m.mutex.RLock() + defer m.mutex.RUnlock() + + msgType := types.MessageType(strings.ToLower(messageType)) + if providers, exists := m.providersByType[msgType]; exists { + return providers + } + return []types.Provider{} +} + +// GetProvidersByMessageType returns all providers grouped by message type +func (m *Service) GetProvidersByMessageType() map[types.MessageType][]types.Provider { + m.mutex.RLock() + defer m.mutex.RUnlock() + + // Create a copy to avoid external modifications + result := make(map[types.MessageType][]types.Provider) + for msgType, providers := range m.providersByType { + result[msgType] = make([]types.Provider, len(providers)) + copy(result[msgType], providers) + } + return result +} + +// GetChannels returns all available channels +func (m *Service) GetChannels() []string { + m.mutex.RLock() + defer m.mutex.RUnlock() + + channels := make([]string, 0, len(m.channels)) + for channel := range m.channels { + channels = append(channels, channel) + } + + // Add default channels + for channel := range m.defaults { + found := false + for _, existing := range channels { + if existing == channel { + found = true + break + } + } + if !found { + channels = append(channels, channel) + } + } + + return channels +} + +// Close closes all provider connections +func (m *Service) Close() error { + m.mutex.Lock() + defer m.mutex.Unlock() + + var errors []string + for name, provider := range m.providers { + if err := provider.Close(); err != nil { + errors = append(errors, fmt.Sprintf("provider %s: %v", name, err)) + } + } + + if len(errors) > 0 { + return fmt.Errorf("close errors: %s", strings.Join(errors, "; ")) + } + return nil +} + +// Helper methods + +// getProviderForChannel returns the provider name for a given channel and message type +func (m *Service) getProviderForChannel(channel, messageType string) string { + // Check channel-specific configuration first + if ch, exists := m.channels[channel]; exists { + if ch.Provider != "" { + return ch.Provider + } + } + + // Check defaults for channel.messageType + key := channel + "." + messageType + if provider, exists := m.defaults[key]; exists { + return provider + } + + // Check defaults for messageType only + if provider, exists := m.defaults[messageType]; exists { + return provider + } + + // Check defaults for channel only + if provider, exists := m.defaults[channel]; exists { + return provider + } + + // If no specific provider configured, try to find any available provider for this message type + msgType := types.MessageType(strings.ToLower(messageType)) + if providers, exists := m.providersByType[msgType]; exists && len(providers) > 0 { + // Return the first available provider (could implement load balancing here) + return providers[0].GetName() + } + + return "" +} + +// parseChannelsConfig parses the channels configuration and converts it to a defaults map +func parseChannelsConfig(channelsConfig map[string]interface{}, defaults map[string]string) { + for channelName, channelData := range channelsConfig { + if channelMap, ok := channelData.(map[string]interface{}); ok { + // Iterate through each message type in the channel + for key, value := range channelMap { + if key == "description" { + // Skip description field + continue + } + + if valueMap, ok := value.(map[string]interface{}); ok { + // This is a message type configuration (email, sms, whatsapp) + if provider, exists := valueMap["provider"]; exists { + if providerStr, ok := provider.(string); ok { + // Set channel.messageType -> provider mapping + defaults[channelName+"."+key] = providerStr + } + } + } else if valueStr, ok := value.(string); ok { + // Direct provider assignment (legacy support) + defaults[channelName+"."+key] = valueStr + } + } + } + } +} + +// validateMessage validates a message before sending +func (m *Service) validateMessage(message *types.Message) error { + if message == nil { + return fmt.Errorf("message is nil") + } + if len(message.To) == 0 { + return fmt.Errorf("message has no recipients") + } + if message.Body == "" && message.HTML == "" { + return fmt.Errorf("message has no content") + } + if message.Type == types.MessageTypeEmail && message.Subject == "" { + return fmt.Errorf("email message requires a subject") + } + return nil +} + +// supportsChannelType checks if a provider supports a given channel type +func (m *Service) supportsChannelType(provider types.Provider, channelType string) bool { + providerType := strings.ToLower(provider.GetType()) + channelType = strings.ToLower(channelType) + + switch channelType { + case "email": + return providerType == "smtp" || providerType == "mailgun" || providerType == "twilio" + case "sms": + return providerType == "twilio" + case "whatsapp": + return providerType == "twilio" + default: + return false + } +} + +// getSupportedMessageTypes returns the message types that a provider supports +func getSupportedMessageTypes(provider types.Provider) []types.MessageType { + providerType := strings.ToLower(provider.GetType()) + + switch providerType { + case "smtp": + return []types.MessageType{types.MessageTypeEmail} + case "mailgun": + return []types.MessageType{types.MessageTypeEmail} + case "twilio": + // Twilio provider supports all message types + return []types.MessageType{types.MessageTypeSMS, types.MessageTypeWhatsApp, types.MessageTypeEmail} + default: + return []types.MessageType{} + } +} diff --git a/messenger/providers/mailgun/mailgun.go b/messenger/providers/mailgun/mailgun.go new file mode 100644 index 00000000..911f24fb --- /dev/null +++ b/messenger/providers/mailgun/mailgun.go @@ -0,0 +1,197 @@ +package mailgun + +import ( + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/yaoapp/yao/messenger/types" +) + +// Provider implements the Provider interface for Mailgun email sending +type Provider struct { + config types.ProviderConfig + domain string + apiKey string + from string + baseURL string + httpClient *http.Client +} + +// NewMailgunProvider creates a new Mailgun provider +func NewMailgunProvider(config types.ProviderConfig) (*Provider, error) { + provider := &Provider{ + config: config, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + } + + // Extract options + options := config.Options + if options == nil { + return nil, fmt.Errorf("Mailgun provider requires options") + } + + // Required options + if domain, ok := options["domain"].(string); ok { + provider.domain = domain + } else { + return nil, fmt.Errorf("Mailgun provider requires 'domain' option") + } + + if apiKey, ok := options["api_key"].(string); ok { + provider.apiKey = apiKey + } else { + return nil, fmt.Errorf("Mailgun provider requires 'api_key' option") + } + + if from, ok := options["from"].(string); ok { + provider.from = from + } else { + return nil, fmt.Errorf("Mailgun provider requires 'from' option") + } + + // Optional options + if baseURL, ok := options["base_url"].(string); ok { + provider.baseURL = baseURL + } else { + // Default to US region + provider.baseURL = "https://api.mailgun.net/v3" + } + + return provider, nil +} + +// Send sends a message using Mailgun +func (p *Provider) Send(message *types.Message) error { + if message.Type != types.MessageTypeEmail { + return fmt.Errorf("Mailgun provider only supports email messages") + } + + return p.sendEmail(message) +} + +// SendBatch sends multiple messages in batch +func (p *Provider) SendBatch(messages []*types.Message) error { + for _, message := range messages { + if err := p.Send(message); err != nil { + return fmt.Errorf("failed to send message to %v: %w", message.To, err) + } + } + return nil +} + +// GetType returns the provider type +func (p *Provider) GetType() string { + return "mailgun" +} + +// GetName returns the provider name +func (p *Provider) GetName() string { + return p.config.Name +} + +// Validate validates the provider configuration +func (p *Provider) Validate() error { + if p.domain == "" { + return fmt.Errorf("domain is required") + } + if p.apiKey == "" { + return fmt.Errorf("api_key is required") + } + if p.from == "" { + return fmt.Errorf("from address is required") + } + return nil +} + +// Close closes the provider connection (no-op for HTTP-based Mailgun) +func (p *Provider) Close() error { + return nil +} + +// sendEmail sends an email via Mailgun API +func (p *Provider) sendEmail(message *types.Message) error { + apiURL := fmt.Sprintf("%s/%s/messages", p.baseURL, p.domain) + + // Prepare form data + data := url.Values{} + + // From address + from := message.From + if from == "" { + from = p.from + } + data.Set("from", from) + + // To addresses + for _, to := range message.To { + data.Add("to", to) + } + + // Subject and content + data.Set("subject", message.Subject) + + if message.Body != "" { + data.Set("text", message.Body) + } + + if message.HTML != "" { + data.Set("html", message.HTML) + } + + // Custom headers + if message.Headers != nil { + for key, value := range message.Headers { + data.Set("h:"+key, value) + } + } + + // Custom variables (metadata) + if message.Metadata != nil { + for key, value := range message.Metadata { + if str, ok := value.(string); ok { + data.Set("v:"+key, str) + } + } + } + + // Priority + if message.Priority > 0 { + data.Set("o:priority", fmt.Sprintf("%d", message.Priority)) + } + + // Scheduled sending + if message.ScheduledAt != nil { + data.Set("o:deliverytime", message.ScheduledAt.Format(time.RFC1123Z)) + } + + // Create request + req, err := http.NewRequest("POST", apiURL, strings.NewReader(data.Encode())) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + // Set authentication + req.SetBasicAuth("api", p.apiKey) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + // Send request + resp, err := p.httpClient.Do(req) + if err != nil { + return fmt.Errorf("failed to send request: %w", err) + } + defer resp.Body.Close() + + // Check response + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("Mailgun API error: %s - %s", resp.Status, string(body)) + } + + return nil +} diff --git a/messenger/providers/smtp/smtp.go b/messenger/providers/smtp/smtp.go new file mode 100644 index 00000000..903357cd --- /dev/null +++ b/messenger/providers/smtp/smtp.go @@ -0,0 +1,286 @@ +package smtp + +import ( + "crypto/tls" + "fmt" + "net/smtp" + "strconv" + "strings" + + "github.com/yaoapp/yao/messenger/types" +) + +// SMTPProvider implements the Provider interface for SMTP email sending +type SMTPProvider struct { + config types.ProviderConfig + host string + port int + username string + password string + from string + useTLS bool + useSSL bool +} + +// NewSMTPProvider creates a new SMTP provider +func NewSMTPProvider(config types.ProviderConfig) (*SMTPProvider, error) { + provider := &SMTPProvider{ + config: config, + useTLS: true, // Default to TLS + } + + // Extract options + options := config.Options + if options == nil { + return nil, fmt.Errorf("SMTP provider requires options") + } + + // Required options + if host, ok := options["host"].(string); ok { + provider.host = host + } else { + return nil, fmt.Errorf("SMTP provider requires 'host' option") + } + + if port, ok := options["port"]; ok { + switch p := port.(type) { + case int: + provider.port = p + case float64: + provider.port = int(p) + case string: + var err error + provider.port, err = strconv.Atoi(p) + if err != nil { + return nil, fmt.Errorf("invalid port: %s", p) + } + default: + return nil, fmt.Errorf("invalid port type") + } + } else { + provider.port = 587 // Default SMTP port + } + + if username, ok := options["username"].(string); ok { + provider.username = username + } else { + return nil, fmt.Errorf("SMTP provider requires 'username' option") + } + + if password, ok := options["password"].(string); ok { + provider.password = password + } else { + return nil, fmt.Errorf("SMTP provider requires 'password' option") + } + + if from, ok := options["from"].(string); ok { + provider.from = from + } else { + return nil, fmt.Errorf("SMTP provider requires 'from' option") + } + + // Optional options + if useTLS, ok := options["use_tls"].(bool); ok { + provider.useTLS = useTLS + } + + if useSSL, ok := options["use_ssl"].(bool); ok { + provider.useSSL = useSSL + } + + return provider, nil +} + +// Send sends a message using SMTP +func (p *SMTPProvider) Send(message *types.Message) error { + if message.Type != types.MessageTypeEmail { + return fmt.Errorf("SMTP provider only supports email messages") + } + + // Create message content + content, err := p.buildMessage(message) + if err != nil { + return fmt.Errorf("failed to build message: %w", err) + } + + // Send the email + return p.sendEmail(message.To, content) +} + +// SendBatch sends multiple messages in batch +func (p *SMTPProvider) SendBatch(messages []*types.Message) error { + for _, message := range messages { + if err := p.Send(message); err != nil { + return fmt.Errorf("failed to send message to %v: %w", message.To, err) + } + } + return nil +} + +// GetType returns the provider type +func (p *SMTPProvider) GetType() string { + return "smtp" +} + +// GetName returns the provider name +func (p *SMTPProvider) GetName() string { + return p.config.Name +} + +// Validate validates the provider configuration +func (p *SMTPProvider) Validate() error { + if p.host == "" { + return fmt.Errorf("host is required") + } + if p.port <= 0 { + return fmt.Errorf("port must be positive") + } + if p.username == "" { + return fmt.Errorf("username is required") + } + if p.password == "" { + return fmt.Errorf("password is required") + } + if p.from == "" { + return fmt.Errorf("from address is required") + } + return nil +} + +// Close closes the provider connection (no-op for SMTP) +func (p *SMTPProvider) Close() error { + return nil +} + +// buildMessage builds the email message content +func (p *SMTPProvider) buildMessage(message *types.Message) (string, error) { + var content strings.Builder + + // From header + from := message.From + if from == "" { + from = p.from + } + content.WriteString(fmt.Sprintf("From: %s\r\n", from)) + + // To header + content.WriteString(fmt.Sprintf("To: %s\r\n", strings.Join(message.To, ", "))) + + // Subject header + content.WriteString(fmt.Sprintf("Subject: %s\r\n", message.Subject)) + + // Additional headers + if message.Headers != nil { + for key, value := range message.Headers { + content.WriteString(fmt.Sprintf("%s: %s\r\n", key, value)) + } + } + + // MIME headers for HTML content + if message.HTML != "" { + content.WriteString("MIME-Version: 1.0\r\n") + if message.Body != "" { + // Multipart message with both text and HTML + content.WriteString("Content-Type: multipart/alternative; boundary=\"boundary123\"\r\n") + content.WriteString("\r\n") + content.WriteString("--boundary123\r\n") + content.WriteString("Content-Type: text/plain; charset=UTF-8\r\n") + content.WriteString("\r\n") + content.WriteString(message.Body) + content.WriteString("\r\n--boundary123\r\n") + content.WriteString("Content-Type: text/html; charset=UTF-8\r\n") + content.WriteString("\r\n") + content.WriteString(message.HTML) + content.WriteString("\r\n--boundary123--\r\n") + } else { + // HTML only + content.WriteString("Content-Type: text/html; charset=UTF-8\r\n") + content.WriteString("\r\n") + content.WriteString(message.HTML) + } + } else { + // Plain text only + content.WriteString("Content-Type: text/plain; charset=UTF-8\r\n") + content.WriteString("\r\n") + content.WriteString(message.Body) + } + + return content.String(), nil +} + +// sendEmail sends the email using SMTP +func (p *SMTPProvider) sendEmail(to []string, content string) error { + addr := fmt.Sprintf("%s:%d", p.host, p.port) + + // Create auth + auth := smtp.PlainAuth("", p.username, p.password, p.host) + + // Send email + if p.useSSL { + // Use SSL/TLS connection + return p.sendWithTLS(addr, auth, to, content) + } else { + // Use standard SMTP with STARTTLS + return smtp.SendMail(addr, auth, p.from, to, []byte(content)) + } +} + +// sendWithTLS sends email with explicit TLS connection +func (p *SMTPProvider) sendWithTLS(addr string, auth smtp.Auth, to []string, content string) error { + // Create TLS connection + tlsConfig := &tls.Config{ + ServerName: p.host, + InsecureSkipVerify: false, + } + + conn, err := tls.Dial("tcp", addr, tlsConfig) + if err != nil { + return fmt.Errorf("failed to create TLS connection: %w", err) + } + defer conn.Close() + + // Create SMTP client + client, err := smtp.NewClient(conn, p.host) + if err != nil { + return fmt.Errorf("failed to create SMTP client: %w", err) + } + defer client.Close() + + // Authenticate + if auth != nil { + if err := client.Auth(auth); err != nil { + return fmt.Errorf("authentication failed: %w", err) + } + } + + // Set sender + if err := client.Mail(p.from); err != nil { + return fmt.Errorf("failed to set sender: %w", err) + } + + // Set recipients + for _, recipient := range to { + if err := client.Rcpt(recipient); err != nil { + return fmt.Errorf("failed to set recipient %s: %w", recipient, err) + } + } + + // Send data + writer, err := client.Data() + if err != nil { + return fmt.Errorf("failed to get data writer: %w", err) + } + + _, err = writer.Write([]byte(content)) + if err != nil { + writer.Close() + return fmt.Errorf("failed to write message: %w", err) + } + + err = writer.Close() + if err != nil { + return fmt.Errorf("failed to close data writer: %w", err) + } + + return client.Quit() +} diff --git a/messenger/providers/twilio/twilio.go b/messenger/providers/twilio/twilio.go new file mode 100644 index 00000000..a3ae4a2e --- /dev/null +++ b/messenger/providers/twilio/twilio.go @@ -0,0 +1,408 @@ +package twilio + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/yaoapp/yao/messenger/types" +) + +// Provider implements the Provider interface for Twilio services (SMS, WhatsApp, Email) +type Provider struct { + config types.ProviderConfig + accountSID string + authToken string + fromPhone string + fromEmail string + fromName string + messagingServiceSID string + sendGridAPIKey string + httpClient *http.Client + baseURL string +} + +// NewTwilioProvider creates a new unified Twilio provider +func NewTwilioProvider(config types.ProviderConfig) (*Provider, error) { + provider := &Provider{ + config: config, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + baseURL: "https://api.twilio.com/2010-04-01", + } + + // Extract options + options := config.Options + if options == nil { + return nil, fmt.Errorf("Twilio provider requires options") + } + + // Required options + if accountSID, ok := options["account_sid"].(string); ok { + provider.accountSID = accountSID + } else { + return nil, fmt.Errorf("Twilio provider requires 'account_sid' option") + } + + if authToken, ok := options["auth_token"].(string); ok { + provider.authToken = authToken + } else { + return nil, fmt.Errorf("Twilio provider requires 'auth_token' option") + } + + // Optional options for different services + if fromPhone, ok := options["from_phone"].(string); ok { + provider.fromPhone = fromPhone + } + + if fromEmail, ok := options["from_email"].(string); ok { + provider.fromEmail = fromEmail + } + + if fromName, ok := options["from_name"].(string); ok { + provider.fromName = fromName + } + + if messagingServiceSID, ok := options["messaging_service_sid"].(string); ok { + provider.messagingServiceSID = messagingServiceSID + } + + if sendGridAPIKey, ok := options["sendgrid_api_key"].(string); ok { + provider.sendGridAPIKey = sendGridAPIKey + } + + if baseURL, ok := options["base_url"].(string); ok { + provider.baseURL = baseURL + } + + return provider, nil +} + +// Send sends a message using appropriate Twilio service based on message type +func (p *Provider) Send(message *types.Message) error { + switch message.Type { + case types.MessageTypeSMS: + return p.sendSMS(message) + case types.MessageTypeWhatsApp: + return p.sendWhatsApp(message) + case types.MessageTypeEmail: + return p.sendEmail(message) + default: + return fmt.Errorf("unsupported message type: %s", message.Type) + } +} + +// SendBatch sends multiple messages in batch +func (p *Provider) SendBatch(messages []*types.Message) error { + for _, message := range messages { + if err := p.Send(message); err != nil { + return fmt.Errorf("failed to send message to %v: %w", message.To, err) + } + } + return nil +} + +// GetType returns the provider type +func (p *Provider) GetType() string { + return "twilio" +} + +// GetName returns the provider name +func (p *Provider) GetName() string { + return p.config.Name +} + +// Validate validates the provider configuration +func (p *Provider) Validate() error { + if p.accountSID == "" { + return fmt.Errorf("account_sid is required") + } + if p.authToken == "" { + return fmt.Errorf("auth_token is required") + } + return nil +} + +// Close closes the provider connection (no-op for HTTP-based Twilio) +func (p *Provider) Close() error { + return nil +} + +// sendSMS sends an SMS message via Twilio +func (p *Provider) sendSMS(message *types.Message) error { + if p.fromPhone == "" && p.messagingServiceSID == "" { + return fmt.Errorf("either from_phone or messaging_service_sid is required for SMS") + } + + for _, to := range message.To { + err := p.sendSMSToRecipient(to, message) + if err != nil { + return fmt.Errorf("failed to send SMS to %s: %w", to, err) + } + } + return nil +} + +// sendSMSToRecipient sends SMS to a single recipient +func (p *Provider) sendSMSToRecipient(to string, message *types.Message) error { + apiURL := fmt.Sprintf("%s/Accounts/%s/Messages.json", p.baseURL, p.accountSID) + + // Prepare form data + data := url.Values{} + data.Set("To", to) + data.Set("Body", message.Body) + + if p.messagingServiceSID != "" { + data.Set("MessagingServiceSid", p.messagingServiceSID) + } else { + data.Set("From", p.fromPhone) + } + + return p.sendTwilioRequest(apiURL, data) +} + +// sendWhatsApp sends a WhatsApp message via Twilio +func (p *Provider) sendWhatsApp(message *types.Message) error { + if p.fromPhone == "" { + return fmt.Errorf("from_phone is required for WhatsApp messages") + } + + for _, to := range message.To { + err := p.sendWhatsAppToRecipient(to, message) + if err != nil { + return fmt.Errorf("failed to send WhatsApp message to %s: %w", to, err) + } + } + return nil +} + +// sendWhatsAppToRecipient sends WhatsApp message to a single recipient +func (p *Provider) sendWhatsAppToRecipient(to string, message *types.Message) error { + apiURL := fmt.Sprintf("%s/Accounts/%s/Messages.json", p.baseURL, p.accountSID) + + // Ensure phone numbers have WhatsApp prefix + fromWhatsApp := p.fromPhone + if !strings.HasPrefix(fromWhatsApp, "whatsapp:") { + fromWhatsApp = "whatsapp:" + fromWhatsApp + } + + toWhatsApp := to + if !strings.HasPrefix(toWhatsApp, "whatsapp:") { + toWhatsApp = "whatsapp:" + toWhatsApp + } + + // Prepare form data + data := url.Values{} + data.Set("From", fromWhatsApp) + data.Set("To", toWhatsApp) + data.Set("Body", message.Body) + + return p.sendTwilioRequest(apiURL, data) +} + +// sendEmail sends an email via Twilio SendGrid API +func (p *Provider) sendEmail(message *types.Message) error { + if p.sendGridAPIKey == "" { + return fmt.Errorf("sendgrid_api_key is required for email messages") + } + if p.fromEmail == "" { + return fmt.Errorf("from_email is required for email messages") + } + + // Create SendGrid email payload + payload := map[string]interface{}{ + "personalizations": []map[string]interface{}{ + { + "to": p.buildEmailRecipients(message.To), + }, + }, + "from": p.buildFromAddress(message), + "subject": message.Subject, + "content": p.buildEmailContent(message), + } + + // Add custom headers if provided + if len(message.Headers) > 0 { + payload["headers"] = message.Headers + } + + // Add attachments if provided + if len(message.Attachments) > 0 { + attachments, err := p.buildAttachments(message.Attachments) + if err != nil { + return fmt.Errorf("failed to build attachments: %w", err) + } + payload["attachments"] = attachments + } + + // Add custom metadata + if len(message.Metadata) > 0 { + customArgs := make(map[string]string) + for key, value := range message.Metadata { + if str, ok := value.(string); ok { + customArgs[key] = str + } + } + if len(customArgs) > 0 { + payload["custom_args"] = customArgs + } + } + + // Add scheduled sending if specified + if message.ScheduledAt != nil { + payload["send_at"] = message.ScheduledAt.Unix() + } + + // Convert to JSON + jsonData, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("failed to marshal email payload: %w", err) + } + + // Send via SendGrid API + apiURL := "https://api.sendgrid.com/v3/mail/send" + req, err := http.NewRequest("POST", apiURL, bytes.NewBuffer(jsonData)) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Authorization", "Bearer "+p.sendGridAPIKey) + req.Header.Set("Content-Type", "application/json") + + resp, err := p.httpClient.Do(req) + if err != nil { + return fmt.Errorf("failed to send request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("SendGrid API error: %s - %s", resp.Status, string(body)) + } + + return nil +} + +// sendTwilioRequest sends a request to Twilio API +func (p *Provider) sendTwilioRequest(apiURL string, data url.Values) error { + // Add custom metadata as status callback parameters + req, err := http.NewRequest("POST", apiURL, strings.NewReader(data.Encode())) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + req.SetBasicAuth(p.accountSID, p.authToken) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + // Send request + resp, err := p.httpClient.Do(req) + if err != nil { + return fmt.Errorf("failed to send request: %w", err) + } + defer resp.Body.Close() + + // Check response + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("Twilio API error: %s - %s", resp.Status, string(body)) + } + + return nil +} + +// buildEmailRecipients builds the recipients array for SendGrid +func (p *Provider) buildEmailRecipients(to []string) []map[string]string { + recipients := make([]map[string]string, len(to)) + for i, email := range to { + recipients[i] = map[string]string{"email": email} + } + return recipients +} + +// buildFromAddress builds the from address for SendGrid +func (p *Provider) buildFromAddress(message *types.Message) map[string]string { + from := map[string]string{ + "email": p.fromEmail, + } + + // Use message from if provided, otherwise use configured from + if message.From != "" { + from["email"] = message.From + } + + // Add name if configured + if p.fromName != "" { + from["name"] = p.fromName + } + + return from +} + +// buildEmailContent builds the content array for SendGrid +func (p *Provider) buildEmailContent(message *types.Message) []map[string]string { + content := []map[string]string{} + + if message.Body != "" { + content = append(content, map[string]string{ + "type": "text/plain", + "value": message.Body, + }) + } + + if message.HTML != "" { + content = append(content, map[string]string{ + "type": "text/html", + "value": message.HTML, + }) + } + + // If no content is provided, use body as plain text + if len(content) == 0 { + content = append(content, map[string]string{ + "type": "text/plain", + "value": "No content provided", + }) + } + + return content +} + +// buildAttachments builds the attachments array for SendGrid +func (p *Provider) buildAttachments(attachments []types.Attachment) ([]map[string]interface{}, error) { + sgAttachments := make([]map[string]interface{}, len(attachments)) + + for i, attachment := range attachments { + // Encode content to base64 + encodedContent := "" + if len(attachment.Content) > 0 { + // Simple base64 encoding (in real implementation, use base64 package) + encodedContent = string(attachment.Content) // This should be base64 encoded + } + + sgAttachment := map[string]interface{}{ + "content": encodedContent, + "filename": attachment.Filename, + "type": attachment.ContentType, + } + + // Add disposition for inline attachments + if attachment.Inline { + sgAttachment["disposition"] = "inline" + if attachment.CID != "" { + sgAttachment["content_id"] = attachment.CID + } + } else { + sgAttachment["disposition"] = "attachment" + } + + sgAttachments[i] = sgAttachment + } + + return sgAttachments, nil +} diff --git a/messenger/types/interfaces.go b/messenger/types/interfaces.go new file mode 100644 index 00000000..9f15073f --- /dev/null +++ b/messenger/types/interfaces.go @@ -0,0 +1,46 @@ +package types + +// Provider defines the interface for message providers +type Provider interface { + // Send sends a message using the provider + Send(message *Message) error + + // SendBatch sends multiple messages in batch + SendBatch(messages []*Message) error + + // GetType returns the provider type (smtp, twilio, mailgun, etc.) + GetType() string + + // GetName returns the provider name/identifier + GetName() string + + // Validate validates the provider configuration + Validate() error + + // Close closes the provider connection if needed + Close() error +} + +// Messenger defines the main messenger interface +type Messenger interface { + // Send sends a message using the specified channel or default provider + Send(channel string, message *Message) error + + // SendWithProvider sends a message using a specific provider + SendWithProvider(providerName string, message *Message) error + + // SendBatch sends multiple messages in batch + SendBatch(channel string, messages []*Message) error + + // GetProvider returns a provider by name + GetProvider(name string) (Provider, error) + + // GetProviders returns all providers for a channel type + GetProviders(channelType string) []Provider + + // GetChannels returns all available channels + GetChannels() []string + + // Close closes all provider connections + Close() error +} diff --git a/messenger/types/types.go b/messenger/types/types.go new file mode 100644 index 00000000..b15ef062 --- /dev/null +++ b/messenger/types/types.go @@ -0,0 +1,119 @@ +package types + +import ( + "time" + + "github.com/yaoapp/gou/types" +) + +// MessageType defines the type of message +type MessageType string + +// Message type constants for different messaging channels +const ( + // MessageTypeEmail represents email messaging + MessageTypeEmail MessageType = "email" + // MessageTypeSMS represents SMS messaging + MessageTypeSMS MessageType = "sms" + // MessageTypeWhatsApp represents WhatsApp messaging + MessageTypeWhatsApp MessageType = "whatsapp" +) + +// Message represents a message to be sent +type Message struct { + Type MessageType `json:"type"` + To []string `json:"to"` + From string `json:"from,omitempty"` + Subject string `json:"subject,omitempty"` // For email + Body string `json:"body"` + HTML string `json:"html,omitempty"` // For email HTML content + Attachments []Attachment `json:"attachments,omitempty"` // For email attachments + Headers map[string]string `json:"headers,omitempty"` // Custom headers + Metadata map[string]interface{} `json:"metadata,omitempty"` // Additional metadata + Priority int `json:"priority,omitempty"` // Message priority + ScheduledAt *time.Time `json:"scheduled_at,omitempty"` // For scheduled sending +} + +// Attachment represents an email attachment +type Attachment struct { + Filename string `json:"filename"` + ContentType string `json:"content_type"` + Content []byte `json:"content"` + Inline bool `json:"inline,omitempty"` // For inline attachments + CID string `json:"cid,omitempty"` // Content-ID for inline attachments +} + +// ProviderConfig represents the configuration for a message provider +type ProviderConfig struct { + types.MetaInfo + Name string `json:"name"` + Description string `json:"description,omitempty"` + Connector string `json:"connector"` // Provider type: smtp, twilio, mailgun + Options map[string]interface{} `json:"options,omitempty"` // Provider-specific options + Enabled bool `json:"enabled,omitempty"` // Whether the provider is enabled (default: true) +} + +// Config represents the messenger configuration +type Config struct { + Defaults map[string]string `json:"defaults,omitempty"` // Default providers for each channel + Channels map[string]Channel `json:"channels,omitempty"` // Channel-specific configurations + Providers []ProviderConfig `json:"providers,omitempty"` // Provider configurations + Global GlobalConfig `json:"global,omitempty"` // Global settings +} + +// Channel represents a message channel configuration +type Channel struct { + Provider string `json:"provider,omitempty"` // Default provider for this channel + Description string `json:"description,omitempty"` // Channel description + Fallbacks []string `json:"fallbacks,omitempty"` // Fallback providers + RateLimit *RateLimit `json:"rate_limit,omitempty"` // Rate limiting settings + Settings map[string]interface{} `json:"settings,omitempty"` // Channel-specific settings + Templates map[string]Template `json:"templates,omitempty"` // Message templates + Types map[string]*Channel `json:"types,omitempty"` // Type-specific configurations (email, sms, whatsapp) +} + +// RateLimit represents rate limiting configuration +type RateLimit struct { + Enabled bool `json:"enabled"` + MaxPerHour int `json:"max_per_hour,omitempty"` + MaxPerDay int `json:"max_per_day,omitempty"` + Window time.Duration `json:"window,omitempty"` +} + +// Template represents a message template +type Template struct { + Subject string `json:"subject,omitempty"` + Body string `json:"body"` + HTML string `json:"html,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` +} + +// GlobalConfig represents global messenger settings +type GlobalConfig struct { + RetryAttempts int `json:"retry_attempts,omitempty"` + RetryDelay time.Duration `json:"retry_delay,omitempty"` + Timeout time.Duration `json:"timeout,omitempty"` + LogLevel string `json:"log_level,omitempty"` +} + +// SendOptions represents options for sending messages +type SendOptions struct { + Provider string `json:"provider,omitempty"` + Template string `json:"template,omitempty"` + Variables map[string]interface{} `json:"variables,omitempty"` + Priority int `json:"priority,omitempty"` + ScheduledAt *time.Time `json:"scheduled_at,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` +} + +// SendResult represents the result of a send operation +type SendResult struct { + Success bool `json:"success"` + MessageID string `json:"message_id,omitempty"` + Provider string `json:"provider"` + Error error `json:"error,omitempty"` + Attempts int `json:"attempts"` + SentAt time.Time `json:"sent_at"` + Metadata map[string]interface{} `json:"metadata,omitempty"` +}