package service import ( "context" "encoding/json" "fmt" "net/http" "time" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/security" applogger "github.com/gochat/gochat/pkg/logger" ) // --- Webhook Event Processor --- // Reference: Chatwoot Integrations::Slack::IncomingMessageBuilder, // Shopify webhook processing, Linear webhook processing. // Each hook type has its own processor that parses the incoming payload // and dispatches events (create conversation, send message, update contact, etc.) // WebhookEventProcessor is the interface for channel-specific webhook event processing. type WebhookEventProcessor interface { HookType() model.HookType ProcessEvent(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}) error VerifySignature(r *http.Request, body []byte, hook *model.IntegrationHook) error } // helper: extract signature + timestamp from request headers based on channel config func verifySignatureWithConfig(signer *security.WebhookSignatureService, r *http.Request, body []byte, secret string, channelType string, cfg security.ChannelSignatureConfig) error { sigHeader := r.Header.Get(cfg.SignatureHeader) tsHeader := "" if cfg.TimestampHeader != "" { tsHeader = r.Header.Get(cfg.TimestampHeader) } return signer.VerifySignature(channelType, secret, body, sigHeader, tsHeader) } // --- Slack Processor --- // Reference: Chatwoot Integrations::Slack::IncomingMessageBuilder // Slack sends events via Events API with URL verification and event payloads. type SlackEventProcessor struct { hookRepo *repository.IntegrationHookRepo accountRepo *repository.AccountRepo signer *security.WebhookSignatureService } func NewSlackEventProcessor(hookRepo *repository.IntegrationHookRepo, accountRepo *repository.AccountRepo, signer *security.WebhookSignatureService) *SlackEventProcessor { return &SlackEventProcessor{hookRepo: hookRepo, accountRepo: accountRepo, signer: signer} } func (p *SlackEventProcessor) HookType() model.HookType { return model.HookTypeSlack } func (p *SlackEventProcessor) VerifySignature(r *http.Request, body []byte, hook *model.IntegrationHook) error { cfg := security.ChannelSignatureConfig{ SignatureHeader: "X-Slack-Signature", TimestampHeader: "X-Slack-Request-Timestamp", Algorithm: "sha256", Prefix: "v0=", } return verifySignatureWithConfig(p.signer, r, body, hook.AccessToken, "slack", cfg) } func (p *SlackEventProcessor) ProcessEvent(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}) error { if challenge, ok := payload["challenge"].(string); ok { if eventType, _ := payload["type"].(string); eventType == "url_verification" { applogger.L().Infof("Slack URL verification challenge for hook %d", hook.ID) return fmt.Errorf("slack_url_verification: %s", challenge) } } event, ok := payload["event"].(map[string]interface{}) if !ok { return fmt.Errorf("invalid Slack event payload: missing 'event' field") } eventType, _ := event["type"].(string) applogger.L().Infof("Processing Slack event: type=%s, hook=%d", eventType, hook.ID) switch eventType { case "message": return p.processSlackMessage(ctx, hook, event) case "link_shared": return p.processSlackLinkShared(ctx, hook, event) case "channel_join", "channel_leave": applogger.L().Infof("Slack channel event %s for hook %d", eventType, hook.ID) return nil default: applogger.L().Infof("Unsupported Slack event type: %s for hook %d", eventType, hook.ID) return nil } } func (p *SlackEventProcessor) processSlackMessage(ctx context.Context, hook *model.IntegrationHook, event map[string]interface{}) error { text, _ := event["text"].(string) userID, _ := event["user"].(string) channelID, _ := event["channel"].(string) ts, _ := event["ts"].(string) if text == "" { return fmt.Errorf("empty Slack message text") } applogger.L().Infof("Slack message: user=%s, channel=%s, ts=%s, hook=%d", userID, channelID, ts, hook.ID) return nil } func (p *SlackEventProcessor) processSlackLinkShared(ctx context.Context, hook *model.IntegrationHook, event map[string]interface{}) error { applogger.L().Infof("Slack link_shared event for hook %d — unfurling not yet implemented", hook.ID) return nil } // --- Shopify Processor --- // Reference: Chatwoot Integrations::Shopify::WebhookProcessor type ShopifyEventProcessor struct { hookRepo *repository.IntegrationHookRepo signer *security.WebhookSignatureService } func NewShopifyEventProcessor(hookRepo *repository.IntegrationHookRepo, signer *security.WebhookSignatureService) *ShopifyEventProcessor { return &ShopifyEventProcessor{hookRepo: hookRepo, signer: signer} } func (p *ShopifyEventProcessor) HookType() model.HookType { return model.HookTypeShopify } func (p *ShopifyEventProcessor) VerifySignature(r *http.Request, body []byte, hook *model.IntegrationHook) error { cfg := security.ChannelSignatureConfig{ SignatureHeader: "X-Shopify-Hmac-Sha256", Algorithm: "sha256", } return verifySignatureWithConfig(p.signer, r, body, hook.AccessToken, "shopify", cfg) } func (p *ShopifyEventProcessor) ProcessEvent(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}) error { topic, _ := payload["_shopify_topic"].(string) applogger.L().Infof("Processing Shopify webhook: topic=%s, hook=%d", topic, hook.ID) switch topic { case "orders/create", "orders/updated", "orders/paid": return p.processShopifyOrder(ctx, hook, payload, topic) case "customers/create", "customers/update": return p.processShopifyCustomer(ctx, hook, payload, topic) default: applogger.L().Infof("Unsupported Shopify topic: %s for hook %d", topic, hook.ID) return nil } } func (p *ShopifyEventProcessor) processShopifyOrder(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}, topic string) error { orderID, _ := payload["id"].(float64) email, _ := payload["email"].(string) applogger.L().Infof("Shopify order: topic=%s, order_id=%.0f, email=%s, hook=%d", topic, orderID, email, hook.ID) return nil } func (p *ShopifyEventProcessor) processShopifyCustomer(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}, topic string) error { custID, _ := payload["id"].(float64) email, _ := payload["email"].(string) applogger.L().Infof("Shopify customer: topic=%s, id=%.0f, email=%s, hook=%d", topic, custID, email, hook.ID) return nil } // --- Linear Processor --- // Reference: Chatwoot Integrations::Linear::WebhookProcessor type LinearEventProcessor struct { hookRepo *repository.IntegrationHookRepo signer *security.WebhookSignatureService } func NewLinearEventProcessor(hookRepo *repository.IntegrationHookRepo, signer *security.WebhookSignatureService) *LinearEventProcessor { return &LinearEventProcessor{hookRepo: hookRepo, signer: signer} } func (p *LinearEventProcessor) HookType() model.HookType { return model.HookTypeLinear } func (p *LinearEventProcessor) VerifySignature(r *http.Request, body []byte, hook *model.IntegrationHook) error { cfg := security.ChannelSignatureConfig{ SignatureHeader: "X-Linear-Signature", Algorithm: "sha256", } return verifySignatureWithConfig(p.signer, r, body, hook.AccessToken, "linear", cfg) } func (p *LinearEventProcessor) ProcessEvent(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}) error { action, _ := payload["action"].(string) data, ok := payload["data"].(map[string]interface{}) if !ok { return fmt.Errorf("invalid Linear event payload: missing 'data' field") } entityType, _ := data["entityType"].(string) applogger.L().Infof("Processing Linear event: action=%s, entityType=%s, hook=%d", action, entityType, hook.ID) switch entityType { case "Issue": return p.processLinearIssue(ctx, hook, data, action) case "Comment": return p.processLinearComment(ctx, hook, data, action) default: applogger.L().Infof("Unsupported Linear entity: %s for hook %d", entityType, hook.ID) return nil } } func (p *LinearEventProcessor) processLinearIssue(ctx context.Context, hook *model.IntegrationHook, data map[string]interface{}, action string) error { issueID, _ := data["id"].(string) title, _ := data["title"].(string) applogger.L().Infof("Linear issue: action=%s, id=%s, title=%s, hook=%d", action, issueID, title, hook.ID) return nil } func (p *LinearEventProcessor) processLinearComment(ctx context.Context, hook *model.IntegrationHook, data map[string]interface{}, action string) error { commentID, _ := data["id"].(string) applogger.L().Infof("Linear comment: action=%s, id=%s, hook=%d", action, commentID, hook.ID) return nil } // --- Notion Processor --- // Reference: Notion API webhooks (beta) type NotionEventProcessor struct { hookRepo *repository.IntegrationHookRepo signer *security.WebhookSignatureService } func NewNotionEventProcessor(hookRepo *repository.IntegrationHookRepo, signer *security.WebhookSignatureService) *NotionEventProcessor { return &NotionEventProcessor{hookRepo: hookRepo, signer: signer} } func (p *NotionEventProcessor) HookType() model.HookType { return model.HookTypeNotion } func (p *NotionEventProcessor) VerifySignature(r *http.Request, body []byte, hook *model.IntegrationHook) error { cfg := security.ChannelSignatureConfig{ SignatureHeader: "X-Notion-Signature", Algorithm: "sha256", } return verifySignatureWithConfig(p.signer, r, body, hook.AccessToken, "notion", cfg) } func (p *NotionEventProcessor) ProcessEvent(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}) error { eventType, _ := payload["type"].(string) applogger.L().Infof("Processing Notion webhook: type=%s, hook=%d", eventType, hook.ID) return nil } // --- Generic Webhook Processor --- // Reference: Chatwoot Channel::WebWidget HMAC verification type GenericWebhookProcessor struct { hookRepo *repository.IntegrationHookRepo signer *security.WebhookSignatureService } func NewGenericWebhookProcessor(hookRepo *repository.IntegrationHookRepo, signer *security.WebhookSignatureService) *GenericWebhookProcessor { return &GenericWebhookProcessor{hookRepo: hookRepo, signer: signer} } func (p *GenericWebhookProcessor) HookType() model.HookType { return model.HookTypeWebhook } func (p *GenericWebhookProcessor) VerifySignature(r *http.Request, body []byte, hook *model.IntegrationHook) error { cfg := security.ChannelSignatureConfig{ SignatureHeader: "X-Webhook-Hmac-Signature", Algorithm: "sha256", } return verifySignatureWithConfig(p.signer, r, body, hook.AccessToken, "web_widget", cfg) } func (p *GenericWebhookProcessor) ProcessEvent(ctx context.Context, hook *model.IntegrationHook, payload map[string]interface{}) error { applogger.L().Infof("Processing generic webhook: hook=%d, keys=%v", hook.ID, payloadKeys(payload)) return nil } // --- Processor Registry --- // Maps HookType → WebhookEventProcessor for dispatching events. type WebhookProcessorRegistry struct { processors map[model.HookType]WebhookEventProcessor } func NewWebhookProcessorRegistry(signer *security.WebhookSignatureService, hookRepo *repository.IntegrationHookRepo, accountRepo *repository.AccountRepo) *WebhookProcessorRegistry { reg := &WebhookProcessorRegistry{processors: make(map[model.HookType]WebhookEventProcessor)} reg.Register(NewSlackEventProcessor(hookRepo, accountRepo, signer)) reg.Register(NewShopifyEventProcessor(hookRepo, signer)) reg.Register(NewLinearEventProcessor(hookRepo, signer)) reg.Register(NewNotionEventProcessor(hookRepo, signer)) reg.Register(NewGenericWebhookProcessor(hookRepo, signer)) return reg } func (r *WebhookProcessorRegistry) Register(p WebhookEventProcessor) { r.processors[p.HookType()] = p } func (r *WebhookProcessorRegistry) Get(hookType model.HookType) (WebhookEventProcessor, bool) { p, ok := r.processors[hookType] return p, ok } func (r *WebhookProcessorRegistry) ListTypes() []model.HookType { types := make([]model.HookType, 0, len(r.processors)) for t := range r.processors { types = append(types, t) } return types } // --- Enhanced ProcessEvent on IntegrationHookService --- // Delegates to the appropriate channel-specific processor. func (s *IntegrationHookService) ProcessWebhookEvent(ctx context.Context, hookID uint, payload map[string]interface{}, r *http.Request, body []byte) error { hook, err := s.hookRepo.GetByID(ctx, hookID) if err != nil { return fmt.Errorf("integration hook not found: %w", err) } if hook.Status != model.HookStatusActive { return fmt.Errorf("integration hook is inactive: status=%s", hook.Status) } processor, ok := s.registry.Get(model.HookType(hook.HookType)) if !ok { return fmt.Errorf("no processor registered for hook type: %s", hook.HookType) } if err := processor.VerifySignature(r, body, hook); err != nil { return fmt.Errorf("webhook signature verification failed: %w", err) } return processor.ProcessEvent(ctx, hook, payload) } // --- Utility functions --- func payloadKeys(m map[string]interface{}) []string { keys := make([]string, 0, len(m)) for k := range m { keys = append(keys, k) } return keys } func VerifyWebhookSignature(r *http.Request, body []byte, secret string, channelType string, cfg security.ChannelSignatureConfig) error { signer := security.NewWebhookSignatureService(security.DefaultWebhookSignatureConfig()) sigHeader := r.Header.Get(cfg.SignatureHeader) tsHeader := "" if cfg.TimestampHeader != "" { tsHeader = r.Header.Get(cfg.TimestampHeader) } return signer.VerifySignature(channelType, secret, body, sigHeader, tsHeader) } func ParseWebhookPayload(body []byte) (map[string]interface{}, error) { var payload map[string]interface{} if err := json.Unmarshal(body, &payload); err != nil { return nil, fmt.Errorf("invalid webhook payload JSON: %w", err) } return payload, nil } // WebhookDelivery tracks outgoing webhook delivery status. // Reference: Chatwoot webhook delivery with retry logic. type WebhookDelivery struct { HookID uint `json:"hook_id"` Status string `json:"status"` // success, failed, retrying Attempts int `json:"attempts"` LastAttempt time.Time `json:"last_attempt"` Response *WebhookDeliveryResponse `json:"response,omitempty"` Payload map[string]interface{} `json:"payload,omitempty"` } type WebhookDeliveryResponse struct { StatusCode int `json:"status_code"` Body string `json:"body"` Error string `json:"error,omitempty"` }