Merge pull request #470 from flanksource/fix/event-consumers

fix: event consumer

events/event_consumer.go

@@ -4,10 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/sethvargo/go-retry"
 	"sync"
 	"time"
 
+	"github.com/sethvargo/go-retry"
+
 	"github.com/flanksource/commons/logger"
 	"github.com/flanksource/incident-commander/api"
 	"github.com/flanksource/incident-commander/db"
@@ -163,7 +164,7 @@ func (e *EventConsumer) listenToPostgresNotify(pgNotify chan bool) {
 }
 
 func (e *EventConsumer) Listen() {
-	logger.Infof("Started listening for database notify events")
+	logger.Infof("Started listening for database notify events: %v", e.WatchEvents)
 
 	if err := e.Validate(); err != nil {
 		logger.Fatalf("Error starting event consumer: %v", err)

events/events.go

@@ -63,13 +63,13 @@ func StartConsumers(gormDB *gorm.DB, config Config) {
 	}
 
 	uniqWatchEvents := set.New[string]()
-	for _, c := range allConsumers {
+	for i, c := range allConsumers {
 		for _, we := range c.WatchEvents {
 			if uniqWatchEvents.Contains(we) {
 				logger.Fatalf("Error starting consumers: event[%s] has multiple consumers", we)
 			}
 		}
 		uniqWatchEvents.Add(c.WatchEvents...)
-		go c.Listen()
+		go allConsumers[i].Listen()
 	}
 }

events/notifications.go

@@ -251,8 +251,8 @@ func addNotificationEvent(ctx *api.Context, event api.Event) error {
 			}
 
 			expressionRunner := pkgNotification.ExpressionRunner{
-				ResourceID:   n.TeamID.String(),
-				ResourceType: "teams",
+				ResourceID:   id,
+				ResourceType: "notification",
 				CelEnv:       celEnv,
 			}