refactor: Implement destination registry (#123)

* refactor: Inverse dependency between models & destinationadapter

* refactor: Rename & update interface of destinationadapter to destregistry

* refactor: Implement destination registry interface

Author: alexluong

internal/deliverymq/messagehandler.go

@@ -8,6 +8,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/hookdeck/outpost/internal/backoff"
 	"github.com/hookdeck/outpost/internal/consumer"
+	"github.com/hookdeck/outpost/internal/destregistry"
 	"github.com/hookdeck/outpost/internal/eventtracer"
 	"github.com/hookdeck/outpost/internal/idempotence"
 	"github.com/hookdeck/outpost/internal/logmq"
@@ -29,6 +30,7 @@ type messageHandler struct {
 	retryBackoff   backoff.Backoff
 	retryMaxCount  int
 	idempotence    idempotence.Idempotence
+	registry       destregistry.Registry
 }
 
 var _ consumer.MessageHandler = (*messageHandler)(nil)
@@ -39,6 +41,7 @@ func NewMessageHandler(
 	logMQ *logmq.LogMQ,
 	entityStore models.EntityStore,
 	logStore models.LogStore,
+	registry destregistry.Registry,
 	eventTracer eventtracer.EventTracer,
 	retryScheduler scheduler.Scheduler,
 	retryBackoff backoff.Backoff,
@@ -50,6 +53,7 @@ func NewMessageHandler(
 		logMQ:          logMQ,
 		entityStore:    entityStore,
 		logStore:       logStore,
+		registry:       registry,
 		retryScheduler: retryScheduler,
 		retryBackoff:   retryBackoff,
 		retryMaxCount:  retryMaxCount,
@@ -103,10 +107,16 @@ func (h *messageHandler) doHandle(ctx context.Context, deliveryEvent models.Deli
 		span.RecordError(errors.New("destination not found"))
 		return err
 	}
-	err = destination.Publish(ctx, &deliveryEvent.Event)
+	provider, err := h.registry.GetProvider(destination.Type)
 	if err != nil {
-		logger.Error("failed to publish event", zap.Error(err))
+		logger.Error("failed to get destination provider", zap.Error(err))
 		span.RecordError(err)
+		return err
+	}
+	var finalErr error
+	if err := provider.Publish(ctx, destination, &deliveryEvent.Event); err != nil {
+		logger.Error("failed to publish event", zap.Error(err))
+		finalErr = err
 		deliveryEvent.Delivery = &models.Delivery{
 			ID:              uuid.New().String(),
 			DeliveryEventID: deliveryEvent.ID,
@@ -127,11 +137,17 @@ func (h *messageHandler) doHandle(ctx context.Context, deliveryEvent models.Deli
 	}
 	logErr := h.logMQ.Publish(ctx, deliveryEvent)
 	if logErr != nil {
-		logger.Error("failed to publish log event", zap.Error(err))
-		span.RecordError(err)
-		err = errors.Join(err, logErr)
+		logger.Error("failed to publish log event", zap.Error(logErr))
+		if finalErr == nil {
+			finalErr = logErr
+		} else {
+			finalErr = errors.Join(finalErr, logErr)
+		}
 	}
-	return err
+	if finalErr != nil {
+		span.RecordError(finalErr)
+	}
+	return finalErr
 }
 
 // shouldRetry checks if the event should be retried.
@@ -146,7 +162,7 @@ func (h *messageHandler) doHandle(ctx context.Context, deliveryEvent models.Deli
 // say logmq.Publish fails. Should that count as an attempt? What about an error BEFORE deliverying the message?
 // Should we write code to differentiate between these two types of errors (predeliveryErr and postdeliveryErr, for example)?
 func (h *messageHandler) shouldRetry(err error, deliveryEvent models.DeliveryEvent) bool {
-	_, isPublishErr := err.(*models.DestinationPublishError)
+	_, isPublishErr := err.(*destregistry.ErrDestinationPublish)
 	if !isPublishErr {
 		return true
 	}

internal/deliverymq/retry_test.go

@@ -79,6 +79,7 @@ func (s *RetryDeliveryMQSuite) SetupTest(t *testing.T) {
 		logMQ,
 		s.entityStore,
 		s.logStore,
+		testutil.Registry,
 		testutil.NewMockEventTracer(s.exporter),
 		retryScheduler,
 		&backoff.ConstantBackoff{Interval: 1 * time.Second},

internal/destinationadapter/adapters/adapter.go

@@ -0,0 +1,21 @@
+package destregistry
+
+import "fmt"
+
+func NewErrDestinationValidation(err error) error {
+	return fmt.Errorf("validation failed: %w", err)
+}
+
+type ErrDestinationPublish struct {
+	Err error
+}
+
+var _ error = &ErrDestinationPublish{}
+
+func (e *ErrDestinationPublish) Error() string {
+	return e.Err.Error()
+}
+
+func NewErrDestinationPublish(err error) error {
+	return &ErrDestinationPublish{Err: err}
+}

internal/destinationadapter/destinationadapter.go

@@ -0,0 +1,14 @@
+package destregistrydefault
+
+import (
+	"github.com/hookdeck/outpost/internal/destregistry"
+	"github.com/hookdeck/outpost/internal/destregistry/providers/destaws"
+	"github.com/hookdeck/outpost/internal/destregistry/providers/destrabbitmq"
+	"github.com/hookdeck/outpost/internal/destregistry/providers/destwebhook"
+)
+
+func RegisterDefault(registry destregistry.Registry) {
+	registry.RegisterProvider("aws", destaws.New())
+	registry.RegisterProvider("rabbitmq", destrabbitmq.New())
+	registry.RegisterProvider("webhook", destwebhook.New())
+}

internal/destregistry/error.go

@@ -1,4 +1,4 @@
-package aws
+package destaws
 
 import (
 	"context"
@@ -11,7 +11,8 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/aws/aws-sdk-go/aws"
-	"github.com/hookdeck/outpost/internal/destinationadapter/adapters"
+	"github.com/hookdeck/outpost/internal/destregistry"
+	"github.com/hookdeck/outpost/internal/models"
 )
 
 type AWSDestination struct {
@@ -28,34 +29,39 @@ type AWSDestinationCredentials struct {
 	Session string // optional
 }
 
-var _ adapters.DestinationAdapter = (*AWSDestination)(nil)
+var _ destregistry.Provider = (*AWSDestination)(nil)
 
 func New() *AWSDestination {
 	return &AWSDestination{}
 }
 
-func (d *AWSDestination) Validate(ctx context.Context, destination adapters.DestinationAdapterValue) error {
+func (d *AWSDestination) Validate(ctx context.Context, destination *models.Destination) error {
 	_, err := parseConfig(destination)
 	if err != nil {
-		return err
+		return destregistry.NewErrDestinationValidation(err)
 	}
-	_, err = parseCredentials(destination)
-	return err
+	if _, err = parseCredentials(destination); err != nil {
+		return destregistry.NewErrDestinationValidation(err)
+	}
+	return nil
 }
 
-func (d *AWSDestination) Publish(ctx context.Context, destination adapters.DestinationAdapterValue, event *adapters.Event) error {
+func (d *AWSDestination) Publish(ctx context.Context, destination *models.Destination, event *models.Event) error {
 	config, err := parseConfig(destination)
 	if err != nil {
-		return err
+		return destregistry.NewErrDestinationPublish(err)
 	}
 	credentials, err := parseCredentials(destination)
 	if err != nil {
-		return err
+		return destregistry.NewErrDestinationPublish(err)
+	}
+	if err := publishEvent(ctx, config, credentials, event); err != nil {
+		return destregistry.NewErrDestinationPublish(err)
 	}
-	return publishEvent(ctx, config, credentials, event)
+	return nil
 }
 
-func parseConfig(destination adapters.DestinationAdapterValue) (*AWSDestinationConfig, error) {
+func parseConfig(destination *models.Destination) (*AWSDestinationConfig, error) {
 	if destination.Type != "aws" {
 		return nil, errors.New("invalid destination type")
 	}
@@ -72,7 +78,7 @@ func parseConfig(destination adapters.DestinationAdapterValue) (*AWSDestinationC
 	return destinationConfig, nil
 }
 
-func parseCredentials(destination adapters.DestinationAdapterValue) (*AWSDestinationCredentials, error) {
+func parseCredentials(destination *models.Destination) (*AWSDestinationCredentials, error) {
 	if destination.Type != "aws" {
 		return nil, errors.New("invalid destination type")
 	}
@@ -94,7 +100,7 @@ func parseCredentials(destination adapters.DestinationAdapterValue) (*AWSDestina
 	return destinationCredentials, nil
 }
 
-func publishEvent(ctx context.Context, cfg *AWSDestinationConfig, creds *AWSDestinationCredentials, event *adapters.Event) error {
+func publishEvent(ctx context.Context, cfg *AWSDestinationConfig, creds *AWSDestinationCredentials, event *models.Event) error {
 	dataBytes, err := json.Marshal(event.Data)
 	if err != nil {
 		return err