feat(notification): add idempotency in notification api

raystack · Dec 16, 2022 · 77988f6 · 77988f6
1 parent ac9f5fc
commit 77988f6
Show file tree

Hide file tree

Showing 34 changed files with 836 additions and 188 deletions.
diff --git a/cli/deps.go b/cli/deps.go
@@ -29,6 +29,7 @@ func InitAPIDeps(
 	encryptor *secret.Crypto,
 	queue notification.Queuer,
 ) (*api.Deps, map[string]notification.Notifier, error) {
+	idempotencyRepository := postgres.NewIdempotencyRepository(pgClient)
 	templateRepository := postgres.NewTemplateRepository(pgClient)
 	templateService := template.NewService(templateRepository)
 
@@ -90,7 +91,7 @@ func InitAPIDeps(
 		receiver.TypeFile:      filePluginService,
 	}
 
-	notificationService := notification.NewService(logger, queue, receiverService, subscriptionService, notifierRegistry)
+	notificationService := notification.NewService(logger, queue, idempotencyRepository, receiverService, subscriptionService, notifierRegistry)
 
 	return &api.Deps{
 			TemplateService:     templateService,

diff --git a/cli/server.go b/cli/server.go
@@ -14,6 +14,7 @@ import (
 	"github.com/odpf/siren/config"
 	"github.com/odpf/siren/core/notification"
 	"github.com/odpf/siren/internal/server"
+	"github.com/odpf/siren/internal/store/postgres"
 	"github.com/odpf/siren/pkg/pgc"
 	"github.com/odpf/siren/pkg/secret"
 	"github.com/odpf/siren/pkg/telemetry"
@@ -113,7 +114,7 @@ func serverMigrateCommand() *cobra.Command {
 				return err
 			}
 
-			if err := pgc.Migrate(cfg.DB); err != nil {
+			if err := postgres.Migrate(cfg.DB); err != nil {
 				return err
 			}
 			printer.Success("Migration done")

diff --git a/core/idempotency/idempotency.go b/core/idempotency/idempotency.go
@@ -0,0 +1,14 @@
+package idempotency
+
+import (
+	"time"
+)
+
+type Idempotency struct {
+	ID        uint64
+	Scope     string
+	Key       string
+	Success   bool
+	CreatedAt time.Time
+	UpdatedAt time.Time
+}
diff --git a/core/notification/handler.go b/core/notification/handler.go
@@ -116,18 +116,20 @@ func (h *Handler) MessageHandler(ctx context.Context, messages []Message) error
 
 		message.MarkPending(time.Now())
 
-		telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessagePending,
+		telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessageCounter,
 			tag.Upsert(telemetry.TagMessageStatus, message.Status.String()),
 			tag.Upsert(telemetry.TagReceiverType, message.ReceiverType))
 
 		newConfig, err := notifier.PostHookQueueTransformConfigs(ctx, message.Configs)
 		if err != nil {
 			message.MarkFailed(time.Now(), false, err)
 
-			telemetry.IncrementInt64Counter(ctx, telemetry.MetricReceiverPostHookQueueFailed,
-				tag.Upsert(telemetry.TagReceiverType, message.ReceiverType))
+			telemetry.IncrementInt64Counter(ctx, telemetry.MetricReceiverHookFailed,
+				tag.Upsert(telemetry.TagReceiverType, message.ReceiverType),
+				tag.Upsert(telemetry.TagHookCondition, telemetry.HookConditionPostHookQueue),
+			)
 
-			telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessageFailed,
+			telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessageCounter,
 				tag.Upsert(telemetry.TagMessageStatus, message.Status.String()),
 				tag.Upsert(telemetry.TagReceiverType, message.ReceiverType))
 
@@ -141,7 +143,7 @@ func (h *Handler) MessageHandler(ctx context.Context, messages []Message) error
 		if retryable, err := notifier.Send(ctx, message); err != nil {
 			message.MarkFailed(time.Now(), retryable, err)
 
-			telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessageFailed,
+			telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessageCounter,
 				tag.Upsert(telemetry.TagMessageStatus, message.Status.String()),
 				tag.Upsert(telemetry.TagReceiverType, message.ReceiverType))
 
@@ -153,7 +155,7 @@ func (h *Handler) MessageHandler(ctx context.Context, messages []Message) error
 
 		message.MarkPublished(time.Now())
 
-		telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessagePublished,
+		telemetry.IncrementInt64Counter(ctx, telemetry.MetricNotificationMessageCounter,
 			tag.Upsert(telemetry.TagMessageStatus, message.Status.String()),
 			tag.Upsert(telemetry.TagReceiverType, message.ReceiverType))
 

diff --git a/core/notification/mocks/idempotency_repository.go b/core/notification/mocks/idempotency_repository.go
diff --git a/core/notification/notification.go b/core/notification/notification.go
@@ -1,11 +1,19 @@
 package notification
 
 import (
+	"context"
 	"time"
 
+	"github.com/odpf/siren/core/idempotency"
 	"github.com/odpf/siren/pkg/errors"
 )
 
+//go:generate mockery --name=IdempotencyRepository -r --case underscore --with-expecter --structname IdempotencyRepository --filename idempotency_repository.go --output=./mocks
+type IdempotencyRepository interface {
+	InsertOnConflictReturning(context.Context, string, string) (*idempotency.Idempotency, error)
+	UpdateSuccess(context.Context, uint64, bool) error
+}
+
 // Notification is a model of notification
 type Notification struct {
 	ID                  string