refactor: event consumer

api/event.go

@@ -8,7 +8,7 @@ import (
 )
 
 type Event struct {
-	ID          uuid.UUID           `json:"id"`
+	ID          uuid.UUID           `gorm:"default:generate_ulid()"`
 	Name        string              `json:"name"`
 	Properties  types.JSONStringMap `json:"properties"`
 	Error       string              `json:"error"`

api/global.go

@@ -67,6 +67,11 @@ func NewContext(db *gorm.DB, echoCtx EchoContext) *Context {
 	return c
 }
 
+func (c Context) WithDB(db *gorm.DB) *Context {
+	c.db = db
+	return &c
+}
+
 func (c *Context) DB() *gorm.DB {
 	if c.db == nil {
 		return nil

api/responders.go

@@ -21,6 +21,7 @@ type IncidentResponders struct {
 
 type Responder struct {
 	ID         uuid.UUID           `json:"id,omitempty"`
+	Type       string              `json:"type"`
 	Properties types.JSONStringMap `json:"properties" gorm:"type:jsonstringmap;<-:false"`
 	ExternalID string              `json:"external_id,omitempty"`
 	IncidentID uuid.UUID           `json:"incident_id,omitempty"`

auth/middleware.go

@@ -61,7 +61,7 @@ func (k *KratosHandler) KratosMiddleware() (*kratosMiddleware, error) {
 	}, nil
 }
 
-var skipAuthPaths = []string{"/health", "/metrics"}
+var skipAuthPaths = []string{"/health", "/metrics", "/kratos/*"}
 
 func canSkipAuth(c echo.Context) bool {
 	return collections.Contains(skipAuthPaths, c.Path())
@@ -102,7 +102,12 @@ func (k *kratosMiddleware) Session(next echo.HandlerFunc) echo.HandlerFunc {
 				email = e
 			}
 		}
-		ctx.WithUser(&api.ContextUser{ID: uuid.MustParse(session.Identity.GetId()), Email: email})
+
+		if uid, err := uuid.Parse(session.Identity.GetId()); err != nil {
+			return c.String(http.StatusUnauthorized, "Unauthorized")
+		} else {
+			ctx.WithUser(&api.ContextUser{ID: uid, Email: email})
+		}
 
 		return next(ctx)
 	}

cmd/server.go

@@ -231,12 +231,11 @@ var Serve = &cobra.Command{
 
 		go jobs.Start()
 
-		events.StartConsumers(db.Gorm, events.Config{
+		events.StartConsumers(db.Gorm, db.Pool, events.Config{
 			UpstreamPush: api.UpstreamConf,
 		})
 
-		playbookRunConsumer := playbook.NewQueueConsumer(db.Gorm, db.Pool)
-		go playbookRunConsumer.Listen()
+		go playbook.StartPlaybookRunConsumer(db.Gorm, db.Pool)
 
 		go launchKopper()
 

db/playbooks.go

@@ -106,17 +106,6 @@ func FindPlaybooksForComponent(ctx *api.Context, configType string, tags map[str
 	return playbooks, err
 }
 
-// GetScheduledPlaybookRuns returns all the scheduled playbook runs that are scheduled to run now
-// or before now.
-func GetScheduledPlaybookRuns(ctx *api.Context, limit int, exceptions ...uuid.UUID) ([]models.PlaybookRun, error) {
-	var runs []models.PlaybookRun
-	if err := ctx.DB().Not(exceptions).Where("start_time <= NOW()").Where("status = ?", models.PlaybookRunStatusScheduled).Limit(limit).Order("start_time").Find(&runs).Error; err != nil {
-		return nil, err
-	}
-
-	return runs, nil
-}
-
 func PersistPlaybookFromCRD(obj *v1.Playbook) error {
 	specJSON, err := json.Marshal(obj.Spec)
 	if err != nil {

events/checks.go

@@ -0,0 +1,8 @@
+package events
+
+func NewCheckConsumerSync() SyncEventConsumer {
+	return SyncEventConsumer{
+		watchEvents: []string{EventCheckPassed, EventCheckFailed},
+		consumers:   []SyncEventHandlerFunc{addNotificationEvent, schedulePlaybookRun},
+	}
+}

events/components.go

@@ -0,0 +1,14 @@
+package events
+
+func NewComponentConsumerSync() SyncEventConsumer {
+	return SyncEventConsumer{
+		watchEvents: []string{
+			EventComponentStatusError,
+			EventComponentStatusHealthy,
+			EventComponentStatusInfo,
+			EventComponentStatusUnhealthy,
+			EventComponentStatusWarning,
+		},
+		consumers: []SyncEventHandlerFunc{addNotificationEvent, schedulePlaybookRun},
+	}
+}