implement slack notifier for demo organization creation event

Author: hvn2k1

api-description/apidocs.swagger.yaml

@@ -3008,6 +3008,7 @@ paths:
               - DOMAIN_EVENT_TAG
               - DOMAIN_EVENT_CODEREF
               - DOMAIN_EVENT_TEAM
+              - DOMAIN_EVENT_DEMO_ORGANIZATION
               - FEATURE_STALE
               - EXPERIMENT_RUNNING
               - MAU_COUNT
@@ -3519,6 +3520,7 @@ definitions:
       - DOMAIN_EVENT_TAG
       - DOMAIN_EVENT_CODEREF
       - DOMAIN_EVENT_TEAM
+      - DOMAIN_EVENT_DEMO_ORGANIZATION
       - FEATURE_STALE
       - EXPERIMENT_RUNNING
       - MAU_COUNT

api-description/web-api.swagger.yaml

@@ -7971,6 +7971,7 @@ paths:
               - DOMAIN_EVENT_TAG
               - DOMAIN_EVENT_CODEREF
               - DOMAIN_EVENT_TEAM
+              - DOMAIN_EVENT_DEMO_ORGANIZATION
               - FEATURE_STALE
               - EXPERIMENT_RUNNING
               - MAU_COUNT
@@ -8699,6 +8700,7 @@ definitions:
       - DOMAIN_EVENT_TAG
       - DOMAIN_EVENT_CODEREF
       - DOMAIN_EVENT_TEAM
+      - DOMAIN_EVENT_DEMO_ORGANIZATION
       - FEATURE_STALE
       - EXPERIMENT_RUNNING
       - MAU_COUNT

manifests/bucketeer/charts/subscriber/values.yaml

@@ -251,6 +251,9 @@ processors:
   userEventPersister:
     flushSize: 200
     flushInterval: 5
+  demoOrganizationCreationNotifier:
+    webURL: ""
+    slackWebhookURL: ""
 
 onDemandProcessors:
   evaluationCountEventDWHPersister:

manifests/bucketeer/values.dev.yaml

@@ -511,6 +511,9 @@ subscriber:
     userEventPersister:
       flushSize: 10
       flushInterval: 5
+    demoOrganizationCreationNotifier:
+      webURL: "https://localhost:9003"
+      slackWebhookURL: ""
 
   onDemandProcessors:
     evaluationCountEventDWHPersister:

pkg/notification/sender/notifier/slack.go

@@ -180,10 +180,42 @@ func (n *slackNotifier) createAttachment(
 		return n.createExperimentRunningAttachment(notification.ExperimentRunningNotification)
 	case sender.Notification_MauCount:
 		return n.createMAUCountAttachment(notification.MauCountNotification)
+	case sender.Notification_DemoOrganizationCreation:
+		return n.createDemoOrganizationCreationAttachment(notification.DemoOrganizationCreationNotification)
 	}
 	return nil, ErrUnknownNotification
 }
 
+func (n *slackNotifier) createDemoOrganizationCreationAttachment(
+	notification *senderproto.DemoOrganizationCreationNotification,
+) (*slack.Attachment, error) {
+	url, err := domainevent.URL(
+		domainproto.Event_ORGANIZATION,
+		n.webURL,
+		"",
+		notification.OrganizationId,
+	)
+	if err != nil {
+		n.logger.Error("Failed to create URL for demo organization",
+			zap.Error(err),
+			zap.String("organizationId", notification.OrganizationId),
+		)
+		return nil, err
+	}
+
+	attachment := &slack.Attachment{
+		Color:      "#36a64f",
+		AuthorName: notification.OwnerEmail,
+		Text: "A new demo organization has been created.\n\n" +
+			"Organization ID: " + notification.OrganizationId + "\n" +
+			"Organization Name: " + notification.OrganizationName + "\n" +
+			"Owner Email: " + notification.OwnerEmail + "\n" +
+			"URL: " + fmt.Sprintf(linkTemplate, url, notification.OrganizationName),
+	}
+
+	return attachment, nil
+}
+
 func (n *slackNotifier) createDomainEventAttachment(
 	notification *senderproto.DomainEventNotification,
 	localizer locale.Localizer,

pkg/subscriber/cmd/server/server.go

@@ -534,7 +534,7 @@ func (s *server) registerPubSubProcessorMap(
 		)
 
 		demoOrganizationCreationNotifier := processor.NewDemoOrganizationCreationNotifier(
-			sender,
+			processorsConfigMap[processor.DemoOrganizationCreationNotifierName],
 			logger,
 		)
 		processors.RegisterProcessor(

pkg/subscriber/processor/demo_organization_creation_notifier.go

@@ -2,31 +2,59 @@ package processor
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"time"
 
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 
-	"github.com/bucketeer-io/bucketeer/pkg/notification/sender"
+	"github.com/bucketeer-io/bucketeer/pkg/notification/sender/notifier"
 	"github.com/bucketeer-io/bucketeer/pkg/pubsub/puller"
 	"github.com/bucketeer-io/bucketeer/pkg/pubsub/puller/codes"
 	"github.com/bucketeer-io/bucketeer/pkg/subscriber"
 	domainevent "github.com/bucketeer-io/bucketeer/proto/event/domain"
 	domaineventproto "github.com/bucketeer-io/bucketeer/proto/event/domain"
+	notificationproto "github.com/bucketeer-io/bucketeer/proto/notification"
+	senderproto "github.com/bucketeer-io/bucketeer/proto/notification/sender"
 )
 
+type DemoOrganizationCreationNotifierConfig struct {
+	WebURL          string `json:"webURL"`
+	SlackWebhookURL string `json:"slackWebhookURL"`
+}
+
 type demoOrganizationCreationNotifier struct {
-	sender sender.Sender
-	logger *zap.Logger
+	slackNotifier                          notifier.Notifier
+	demoOrganizationCreationNotifierConfig DemoOrganizationCreationNotifierConfig
+	logger                                 *zap.Logger
 }
 
 func NewDemoOrganizationCreationNotifier(
-	sender sender.Sender,
+	config interface{},
 	logger *zap.Logger,
 ) subscriber.PubSubProcessor {
+	jsonConfigMap, ok := config.(map[string]interface{})
+	if !ok {
+		logger.Error("demoOrganizationCreationNotifier: invalid config type, expected map[string]interface{}")
+		return nil
+	}
+	configBytes, err := json.Marshal(jsonConfigMap)
+	if err != nil {
+		logger.Error("demoOrganizationCreationNotifier: failed to marshal config", zap.Error(err))
+		return nil
+	}
+	var notifierConfig DemoOrganizationCreationNotifierConfig
+	if err := json.Unmarshal(configBytes, &notifierConfig); err != nil {
+		logger.Error("demoOrganizationCreationNotifier: failed to unmarshal config", zap.Error(err))
+		return nil
+	}
+	slackNotifier := notifier.NewSlackNotifier(notifierConfig.WebURL)
+
 	return &demoOrganizationCreationNotifier{
-		sender: sender,
-		logger: logger,
+		slackNotifier:                          slackNotifier,
+		demoOrganizationCreationNotifierConfig: notifierConfig,
+		logger:                                 logger,
 	}
 }
 
@@ -55,6 +83,11 @@ func (d demoOrganizationCreationNotifier) handleMessage(msg *puller.Message) {
 	}
 	domainEvent, err := d.unmarshalMessage(msg)
 	if err != nil {
+		d.logger.Error("Failed to unmarshal message",
+			zap.Error(err),
+			zap.String("msgID", msg.ID),
+			zap.String("attributes", fmt.Sprintf("%+v", msg.Attributes)),
+		)
 		subscriberHandledCounter.WithLabelValues(subscriberDemoOrganizationEvent, codes.BadMessage.String()).Inc()
 		msg.Ack()
 		return
@@ -67,11 +100,42 @@ func (d demoOrganizationCreationNotifier) handleMessage(msg *puller.Message) {
 		return
 	}
 
-	err = d.SendSlackNotifier(ctx, domainEvent)
+	var organizationCreatedEvent domaineventproto.OrganizationCreatedEvent
+	if err := domainEvent.Data.UnmarshalTo(&organizationCreatedEvent); err != nil {
+		d.logger.Error("Failed to unmarshal OrganizationCreatedEvent",
+			zap.String("event id", domainEvent.Id),
+			zap.Error(err),
+		)
+		subscriberHandledCounter.WithLabelValues(
+			subscriberDemoOrganizationEvent,
+			codes.NonRepeatableError.String(),
+		).Inc()
+		msg.Ack()
+		return
+	}
+
+	recipient := &notificationproto.Recipient{
+		Type:     notificationproto.Recipient_SlackChannel,
+		Language: notificationproto.Recipient_ENGLISH,
+		SlackChannelRecipient: &notificationproto.SlackChannelRecipient{
+			WebhookUrl: d.demoOrganizationCreationNotifierConfig.SlackWebhookURL,
+		},
+	}
+	fmt.Printf("?%+v\n", recipient)
+	err = d.slackNotifier.Notify(ctx, &senderproto.Notification{
+		Type: senderproto.Notification_DemoOrganizationCreation,
+		DemoOrganizationCreationNotification: &senderproto.DemoOrganizationCreationNotification{
+			OwnerEmail:       organizationCreatedEvent.OwnerEmail,
+			OrganizationId:   organizationCreatedEvent.Id,
+			OrganizationName: organizationCreatedEvent.Name,
+		},
+	}, recipient, recipient.Language)
 	if err != nil {
-		d.logger.Error("Failed to send Slack notification",
+		d.logger.Error("Failed to send notification",
 			zap.Error(err),
-			zap.String("eventID", domainEvent.Id),
+			zap.String("event id", domainEvent.Id),
+			zap.String("webhookURL", d.demoOrganizationCreationNotifierConfig.SlackWebhookURL),
+			zap.String("organizationId", organizationCreatedEvent.Id),
 		)
 		subscriberHandledCounter.WithLabelValues(
 			subscriberDemoOrganizationEvent,
@@ -80,16 +144,14 @@ func (d demoOrganizationCreationNotifier) handleMessage(msg *puller.Message) {
 		msg.Ack()
 		return
 	}
+	fmt.Printf("?2")
+	subscriberHandledCounter.WithLabelValues(
+		subscriberDemoOrganizationEvent,
+		codes.OK.String(),
+	).Inc()
 	msg.Ack()
 }
 
-func (d demoOrganizationCreationNotifier) SendSlackNotifier(
-	ctx context.Context,
-	event *domainevent.Event,
-) error {
-
-}
-
 func (d demoOrganizationCreationNotifier) unmarshalMessage(msg *puller.Message) (*domainevent.Event, error) {
 	event := &domaineventproto.Event{}
 	err := proto.Unmarshal(msg.Data, event)