package service import ( "context" "encoding/json" "fmt" "github.com/ThreeDotsLabs/watermill" "github.com/ThreeDotsLabs/watermill-redisstream/pkg/redisstream" "github.com/ThreeDotsLabs/watermill/message" "github.com/redis/go-redis/v9" applogger "github.com/gochat/gochat/pkg/logger" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/pubsub" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/security" ) // NotificationDeliveryService bridges Watermill PubSub events to notification // creation and multi-channel delivery (push, email, webhook). // // Architecture mapping: // Chatwoot Wisper → Sidekiq workers → NotificationDeliveryService // Each event (message_created, conversation_assigned, etc.) triggers: // 1. Create a Notification record in the DB // 2. Check NotificationPreference for push/email enabled // 3. Deliver push via PushDeliveryService // 4. Deliver webhook via WebhookDeliveryService (HTTP POST with HMAC signing) // // Reference: Chatwoot notification_service.rb + P2B M8 spec type NotificationDeliveryService struct { notificationService *NotificationService pushDeliveryService *PushDeliveryService webhookDeliverySvc *WebhookDeliveryService webhookSignatureSvc *security.WebhookSignatureService notifPrefRepo *repository.NotificationPreferenceRepo pushTokenRepo *repository.PushTokenRepo webhookSubRepo *repository.WebhookSubscriptionRepo router *message.Router subscriber *redisstream.Subscriber } // NewNotificationDeliveryService creates a delivery service and registers Watermill handlers. func NewNotificationDeliveryService( notificationService *NotificationService, pushDeliveryService *PushDeliveryService, webhookDeliverySvc *WebhookDeliveryService, webhookSignatureSvc *security.WebhookSignatureService, notifPrefRepo *repository.NotificationPreferenceRepo, pushTokenRepo *repository.PushTokenRepo, webhookSubRepo *repository.WebhookSubscriptionRepo, redisClient redis.UniversalClient, ) (*NotificationDeliveryService, error) { loggerAdapter := &deliveryWatermillAdapter{} subscriber, err := redisstream.NewSubscriber( redisstream.SubscriberConfig{ Client: redisClient, ConsumerGroup: "notif-delivery-" + watermill.NewUUID(), }, loggerAdapter, ) if err != nil { return nil, fmt.Errorf("failed to create delivery subscriber: %w", err) } router, err := message.NewRouter(message.RouterConfig{}, loggerAdapter) if err != nil { return nil, fmt.Errorf("failed to create delivery router: %w", err) } s := &NotificationDeliveryService{ notificationService: notificationService, pushDeliveryService: pushDeliveryService, webhookDeliverySvc: webhookDeliverySvc, webhookSignatureSvc: webhookSignatureSvc, notifPrefRepo: notifPrefRepo, pushTokenRepo: pushTokenRepo, webhookSubRepo: webhookSubRepo, router: router, subscriber: subscriber, } s.registerHandlers() return s, nil } // --- Notification event type mapping --- // Maps Watermill topic names to Notification.notification_type values. var topicToNotificationType = map[string]string{ pubsub.TopicMessageCreated: "message_created", pubsub.TopicMessageUpdated: "message_updated", pubsub.TopicMessageDeleted: "message_deleted", pubsub.TopicConversationCreated: "conversation_created", pubsub.TopicConversationUpdated: "conversation_updated", pubsub.TopicConversationResolved: "conversation_resolved", pubsub.TopicConversationAssigned: "conversation_assigned", pubsub.TopicContactCreated: "contact_created", pubsub.TopicContactUpdated: "contact_updated", pubsub.TopicInboxCreated: "inbox_created", pubsub.TopicInboxUpdated: "inbox_updated", pubsub.TopicInboxDeleted: "inbox_deleted", } // --- Handler registration --- // Each handler listens on a Watermill topic, creates a notification record, // and triggers push/webhook delivery if enabled. func (s *NotificationDeliveryService) registerHandlers() { for topic, notifType := range topicToNotificationType { handlerName := "notif-delivery-" + notifType + "-handler" s.router.AddNoPublisherHandler( handlerName, topic, s.subscriber, s.handleNotificationEvent(notifType), ) } // System notification topic — special handling s.router.AddNoPublisherHandler( "notif-delivery-system-notification-handler", pubsub.TopicSystemNotification, s.subscriber, s.handleSystemNotification(), ) } // --- Event payload structure --- // The Watermill message payload is expected to be JSON with these fields. type notificationEventPayload struct { AccountID uint `json:"account_id"` UserID uint `json:"user_id"` ConversationID uint `json:"conversation_id,omitempty"` PrimaryActorType string `json:"primary_actor_type,omitempty"` PrimaryActorID uint `json:"primary_actor_id,omitempty"` SecondaryActorType string `json:"secondary_actor_type,omitempty"` SecondaryActorID uint `json:"secondary_actor_id,omitempty"` PushEnabled bool `json:"push_enabled,omitempty"` EmailEnabled bool `json:"email_enabled,omitempty"` AdditionalAttributes json.RawMessage `json:"additional_attributes,omitempty"` } // handleNotificationEvent creates a handler for a specific notification type. func (s *NotificationDeliveryService) handleNotificationEvent(notifType string) func(msg *message.Message) error { return func(msg *message.Message) error { var payload notificationEventPayload if err := json.Unmarshal(msg.Payload, &payload); err != nil { applogger.L().Errorf("notif-delivery: failed to unmarshal payload for %s: %v", notifType, err) return nil // don't retry — bad payload } if payload.UserID == 0 { applogger.L().Warnf("notif-delivery: event %s missing user_id, skipping", notifType) return nil } ctx := context.Background() // Step 1: Create notification record notif := &model.Notification{ AccountID: nilIfZero(payload.AccountID), UserID: payload.UserID, NotificationType: notifType, PrimaryActorType: payload.PrimaryActorType, PrimaryActorID: payload.PrimaryActorID, SecondaryActorType: payload.SecondaryActorType, SecondaryActorID: payload.SecondaryActorID, PushEnabled: payload.PushEnabled, EmailEnabled: payload.EmailEnabled, AdditionalAttributes: payload.AdditionalAttributes, } if err := s.notificationService.CreateNotification(ctx, notif); err != nil { applogger.L().Errorf("notif-delivery: failed to create notification for %s: %v", notifType, err) return nil // don't retry — DB error } applogger.L().Infof("notif-delivery: created notification id=%d type=%s for user=%d", notif.ID, notifType, payload.UserID) // Step 2: Check notification preferences for push delivery s.deliverToChannels(ctx, notif, payload) // Step 3: Deliver to webhook subscriptions if payload.AccountID > 0 { s.deliverToWebhooks(ctx, payload.AccountID, notifType, msg.Payload) } return nil } } // handleSystemNotification handles system-wide notification events. func (s *NotificationDeliveryService) handleSystemNotification() func(msg *message.Message) error { return func(msg *message.Message) error { var payload notificationEventPayload if err := json.Unmarshal(msg.Payload, &payload); err != nil { applogger.L().Errorf("notif-delivery: failed to unmarshal system notification: %v", err) return nil } if payload.UserID == 0 { applogger.L().Warnf("notif-delivery: system notification missing user_id, skipping") return nil } ctx := context.Background() notif := &model.Notification{ AccountID: nilIfZero(payload.AccountID), UserID: payload.UserID, NotificationType: "system_notification", PrimaryActorType: payload.PrimaryActorType, PrimaryActorID: payload.PrimaryActorID, AdditionalAttributes: payload.AdditionalAttributes, } if err := s.notificationService.CreateNotification(ctx, notif); err != nil { applogger.L().Errorf("notif-delivery: failed to create system notification: %v", err) } return nil } } // deliverToChannels checks preferences and delivers via push. func (s *NotificationDeliveryService) deliverToChannels(ctx context.Context, notif *model.Notification, payload notificationEventPayload) { if !payload.PushEnabled || s.pushDeliveryService == nil { return } // Check push preferences var accountID uint if notif.AccountID != nil { accountID = *notif.AccountID } prefs, err := s.notifPrefRepo.ListByUserAndAccount(ctx, payload.UserID, accountID) if err != nil { applogger.L().Errorf("notif-delivery: failed to fetch push preferences: %v", err) return } pushEnabled := isPushEnabled(prefs, notif.NotificationType) if !pushEnabled { applogger.L().Debugf("notif-delivery: push disabled for user=%d event=%s", payload.UserID, notif.NotificationType) return } pushPayload := PushPayload{ Title: notif.NotificationType, Body: fmt.Sprintf("New %s notification", notif.NotificationType), Data: map[string]interface{}{ "notification_id": notif.ID, "notification_type": notif.NotificationType, "account_id": payload.AccountID, "conversation_id": payload.ConversationID, }, } if err := s.pushDeliveryService.SendPushNotification(ctx, payload.UserID, pushPayload); err != nil { applogger.L().Errorf("notif-delivery: push delivery failed for user=%d: %v", payload.UserID, err) } } // deliverToWebhooks finds matching webhook subscriptions and delivers via HTTP POST. func (s *NotificationDeliveryService) deliverToWebhooks(ctx context.Context, accountID uint, eventType string, payload []byte) { if s.webhookDeliverySvc == nil { return } // Convert raw JSON payload to map for WebhookDeliveryService.DeliverEvent var payloadMap map[string]interface{} if err := json.Unmarshal(payload, &payloadMap); err != nil { applogger.L().Errorf("notif-delivery: failed to unmarshal webhook payload: %v", err) return } if err := s.webhookDeliverySvc.DeliverEvent(ctx, accountID, eventType, payloadMap); err != nil { applogger.L().Errorf("notif-delivery: webhook delivery failed for account=%d event=%s: %v", accountID, eventType, err) } } // isPushEnabled checks if push is enabled for a given event type based on preferences. func isPushEnabled(prefs []model.NotificationPreference, eventType string) bool { for _, p := range prefs { if p.Channel == "push" && p.EventType == eventType { return p.Enabled } } // Default: push is enabled if no preference is set return true } // nilIfZero returns nil pointer for zero value, otherwise a pointer to the value. func nilIfZero(v uint) *uint { if v == 0 { return nil } return &v } // Start runs the Watermill router (begins consuming events). func (s *NotificationDeliveryService) Start(ctx context.Context) error { applogger.L().Info("notif-delivery: starting notification delivery router") return s.router.Run(ctx) } // Close shuts down the delivery router and subscriber. func (s *NotificationDeliveryService) Close() error { if err := s.router.Close(); err != nil { applogger.L().Errorf("notif-delivery: failed to close router: %v", err) } return s.subscriber.Close() } // --- Watermill logger adapter for delivery service --- type deliveryWatermillAdapter struct{} func (a *deliveryWatermillAdapter) Error(msg string, err error, fields watermill.LogFields) { applogger.L().Errorf("notif-delivery watermill: %s err=%v fields=%v", msg, err, fields) } func (a *deliveryWatermillAdapter) Info(msg string, fields watermill.LogFields) { applogger.L().Infof("notif-delivery watermill: %s fields=%v", msg, fields) } func (a *deliveryWatermillAdapter) Debug(msg string, fields watermill.LogFields) { // Suppress debug noise } func (a *deliveryWatermillAdapter) Trace(msg string, fields watermill.LogFields) { // Suppress trace noise } func (a *deliveryWatermillAdapter) With(fields watermill.LogFields) watermill.LoggerAdapter { return a }