Fix webhook selection code

Author: AchoArnold

api/pkg/listeners/webhook_listener.go

@@ -46,7 +46,7 @@ func (listener *WebhookListener) OnMessagePhoneReceived(ctx context.Context, eve
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
 	}
 
-	if err := listener.service.Send(ctx, payload.UserID, event); err != nil {
+	if err := listener.service.Send(ctx, payload.UserID, event, payload.Owner); err != nil {
 		msg := fmt.Sprintf("cannot process [%s] event with ID [%s]", event.Type(), event.ID())
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))
 	}

api/pkg/repositories/gorm_webhook_repository.go

@@ -64,12 +64,15 @@ func (repository *gormWebhookRepository) Index(ctx context.Context, userID entit
 	return webhooks, nil
 }
 
-func (repository *gormWebhookRepository) LoadByEvent(ctx context.Context, userID entities.UserID, event string) ([]*entities.Webhook, error) {
+func (repository *gormWebhookRepository) LoadByEvent(ctx context.Context, userID entities.UserID, event string, phoneNumber string) ([]*entities.Webhook, error) {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
 
 	webhooks := make([]*entities.Webhook, 0)
-	err := repository.db.Raw("SELECT * FROM webhooks WHERE user_id = ? AND CAST(? as TEXT) = ANY(events)", userID, event).Scan(&webhooks).Error
+	err := repository.db.
+		Raw("SELECT * FROM webhooks WHERE user_id = ? AND CAST(? as TEXT) = ANY(events) AND CAST(? as TEXT) = ANY(phone_numbers)", userID, event, phoneNumber).
+		Scan(&webhooks).
+		Error
 	if err != nil {
 		msg := fmt.Sprintf("cannot load webhooks for user with ID [%s] and event [%s]", userID, event)
 		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, msg))

api/pkg/repositories/webhook_repository.go

@@ -17,7 +17,7 @@ type WebhookRepository interface {
 	Index(ctx context.Context, userID entities.UserID, params IndexParams) ([]*entities.Webhook, error)
 
 	// LoadByEvent loads webhooks for a user and event.
-	LoadByEvent(ctx context.Context, userID entities.UserID, event string) ([]*entities.Webhook, error)
+	LoadByEvent(ctx context.Context, userID entities.UserID, event string, phoneNumber string) ([]*entities.Webhook, error)
 
 	// Load loads a webhook by ID.
 	Load(ctx context.Context, userID entities.UserID, webhookID uuid.UUID) (*entities.Webhook, error)

api/pkg/services/webhook_service.go

@@ -157,11 +157,11 @@ func (service *WebhookService) Update(ctx context.Context, params *WebhookUpdate
 }
 
 // Send an event to a subscribed webhook
-func (service *WebhookService) Send(ctx context.Context, userID entities.UserID, event cloudevents.Event) error {
+func (service *WebhookService) Send(ctx context.Context, userID entities.UserID, event cloudevents.Event, phoneNumber string) error {
 	ctx, span, ctxLogger := service.tracer.StartWithLogger(ctx, service.logger)
 	defer span.End()
 
-	webhooks, err := service.repository.LoadByEvent(ctx, userID, event.Type())
+	webhooks, err := service.repository.LoadByEvent(ctx, userID, event.Type(), phoneNumber)
 	if err != nil {
 		msg := fmt.Sprintf("cannot load webhooks for userID [%s] and event [%s]", userID, event.Type())
 		return service.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCode(err, stacktrace.GetCode(err), msg))