feat: allow system config changes (#3876)

* feat: run repeatable setup steps

* feat: react to system config changes

* renaming

internal/migration/command.go

@@ -15,20 +15,23 @@ import (
 type SetupStep struct {
 	eventstore.BaseEvent `json:"-"`
 	migration            Migration
-	Name                 string `json:"name"`
-	Error                error  `json:"error,omitempty"`
+	Name                 string      `json:"name"`
+	Error                error       `json:"error,omitempty"`
+	LastRun              interface{} `json:"lastRun,omitempty"`
 }
 
 func (s *SetupStep) UnmarshalJSON(data []byte) error {
 	fields := struct {
-		Name  string            `json:"name,"`
-		Error *errors.CaosError `json:"error"`
+		Name    string                 `json:"name,"`
+		Error   *errors.CaosError      `json:"error"`
+		LastRun map[string]interface{} `json:"lastRun,omitempty"`
 	}{}
 	if err := json.Unmarshal(data, &fields); err != nil {
 		return err
 	}
 	s.Name = fields.Name
 	s.Error = fields.Error
+	s.LastRun = fields.LastRun
 	return nil
 }
 
@@ -46,15 +49,21 @@ func setupStartedCmd(migration Migration) eventstore.Command {
 
 func setupDoneCmd(migration Migration, err error) eventstore.Command {
 	ctx := authz.SetCtxData(service.WithService(context.Background(), "system"), authz.CtxData{UserID: "system", OrgID: "SYSTEM", ResourceOwner: "SYSTEM"})
+	typ := doneType
+	var lastRun interface{}
+	if repeatable, ok := migration.(RepeatableMigration); ok {
+		typ = repeatableDoneType
+		lastRun = repeatable
+	}
+	if err != nil {
+		typ = failedType
+	}
+
 	s := &SetupStep{
 		migration: migration,
 		Name:      migration.String(),
 		Error:     err,
-	}
-
-	typ := doneType
-	if err != nil {
-		typ = failedType
+		LastRun:   lastRun,
 	}
 
 	s.BaseEvent = *eventstore.NewBaseEventForPush(
@@ -75,7 +84,8 @@ func (s *SetupStep) UniqueConstraints() []*eventstore.EventUniqueConstraint {
 		return []*eventstore.EventUniqueConstraint{
 			eventstore.NewAddGlobalEventUniqueConstraint("migration_started", s.migration.String(), "Errors.Step.Started.AlreadyExists"),
 		}
-	case failedType:
+	case failedType,
+		repeatableDoneType:
 		return []*eventstore.EventUniqueConstraint{
 			eventstore.NewRemoveGlobalEventUniqueConstraint("migration_started", s.migration.String()),
 		}
@@ -90,6 +100,7 @@ func RegisterMappers(es *eventstore.Eventstore) {
 	es.RegisterFilterEventMapper(startedType, SetupMapper)
 	es.RegisterFilterEventMapper(doneType, SetupMapper)
 	es.RegisterFilterEventMapper(failedType, SetupMapper)
+	es.RegisterFilterEventMapper(repeatableDoneType, SetupMapper)
 }
 
 func SetupMapper(event *repository.Event) (eventstore.Event, error) {

internal/migration/migration.go

@@ -10,11 +10,12 @@ import (
 )
 
 const (
-	startedType   = eventstore.EventType("system.migration.started")
-	doneType      = eventstore.EventType("system.migration.done")
-	failedType    = eventstore.EventType("system.migration.failed")
-	aggregateType = eventstore.AggregateType("system")
-	aggregateID   = "SYSTEM"
+	startedType        = eventstore.EventType("system.migration.started")
+	doneType           = eventstore.EventType("system.migration.done")
+	failedType         = eventstore.EventType("system.migration.failed")
+	repeatableDoneType = eventstore.EventType("system.migration.repeatable.done")
+	aggregateType      = eventstore.AggregateType("system")
+	aggregateID        = "SYSTEM"
 )
 
 type Migration interface {
@@ -22,7 +23,15 @@ type Migration interface {
 	Execute(context.Context) error
 }
 
+type RepeatableMigration interface {
+	Migration
+	SetLastExecution(lastRun map[string]interface{})
+	Check() bool
+}
+
 func Migrate(ctx context.Context, es *eventstore.Eventstore, migration Migration) (err error) {
+	logging.Infof("verify migration %s", migration.String())
+
 	if should, err := shouldExec(ctx, es, migration); !should || err != nil {
 		return err
 	}
@@ -31,6 +40,7 @@ func Migrate(ctx context.Context, es *eventstore.Eventstore, migration Migration
 		return err
 	}
 
+	logging.Infof("starting migration %s", migration.String())
 	err = migration.Execute(ctx)
 	logging.OnError(err).Error("migration failed")
 
@@ -48,7 +58,7 @@ func shouldExec(ctx context.Context, es *eventstore.Eventstore, migration Migrat
 		AddQuery().
 		AggregateTypes(aggregateType).
 		AggregateIDs(aggregateID).
-		EventTypes(startedType, doneType, failedType).
+		EventTypes(startedType, doneType, repeatableDoneType, failedType).
 		Builder())
 	if err != nil {
 		return false, err
@@ -68,10 +78,23 @@ func shouldExec(ctx context.Context, es *eventstore.Eventstore, migration Migrat
 		switch event.Type() {
 		case startedType, failedType:
 			isStarted = !isStarted
-		case doneType:
-			return false, nil
+		case doneType,
+			repeatableDoneType:
+			repeatable, ok := migration.(RepeatableMigration)
+			if !ok {
+				return false, nil
+			}
+			isStarted = false
+			repeatable.SetLastExecution(e.LastRun.(map[string]interface{}))
 		}
 	}
 
-	return !isStarted, nil
+	if isStarted {
+		return false, nil
+	}
+	repeatable, ok := migration.(RepeatableMigration)
+	if !ok {
+		return true, nil
+	}
+	return repeatable.Check(), nil
 }