Extract some duplicated code snippets

Author: yhabteab

pkg/icingadb/runtime_updates.go

@@ -203,32 +203,8 @@ func (r *RuntimeUpdates) Sync(
 			// Start the regular Icinga DB checkables upsert stream.
 			g.Go(upsertEntityFunc(entities))
 
-			deletedIds := make(chan any, r.redis.Options.XReadCount)
-			g.Go(func() error {
-				defer close(deletedIds)
-
-				var counter com.Counter
-				defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
-					if count := counter.Reset(); count > 0 {
-						r.logger.Infof("Updating %d %s sla lifecycles", count, s.Name())
-					}
-				}).Stop()
-
-				sl := v1.NewSlaLifecycle()
-				stmt := fmt.Sprintf(`UPDATE %s SET delete_time = :delete_time WHERE "id" = :id AND "delete_time" = 0`, database.TableName(sl))
-				sem := r.db.GetSemaphoreForTable(database.TableName(sl))
-
-				// extractEntityId is used as a callback for the on success mechanism to extract the checkables id.
-				extractEntityId := func(e database.Entity) any { return e.(*v1.SlaLifecycle).SourceEntity.ID() }
-
-				return r.db.NamedBulkExec(
-					ctx, stmt, upsertCount, sem, CreateSlaLifecyclesFromCheckables(ctx, g, deleteEntities, true),
-					com.NeverSplit[database.Entity], database.OnSuccessIncrement[database.Entity](&counter),
-					OnSuccessApplyAndSendTo[database.Entity, any](deletedIds, extractEntityId))
-			})
-
 			// Start the regular Icinga DB checkables delete stream.
-			g.Go(deleteEntityFunc(deletedIds))
+			g.Go(deleteEntityFunc(StreamIDsFromUpdatedSlaLifecycles(ctx, r.db, g, r.logger, deleteEntities, upsertCount)))
 		default:
 			// For non-checkables runtime updates of upsert event
 			g.Go(upsertEntityFunc(upsertEntities))

pkg/icingadb/sla_lifecycle.go

@@ -2,14 +2,21 @@ package icingadb
 
 import (
 	"context"
+	"fmt"
+	"github.com/icinga/icinga-go-library/com"
 	"github.com/icinga/icinga-go-library/database"
+	"github.com/icinga/icinga-go-library/logging"
+	"github.com/icinga/icinga-go-library/periodic"
 	"github.com/icinga/icinga-go-library/types"
 	v1 "github.com/icinga/icingadb/pkg/icingadb/v1"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 	"time"
 )
 
+// tableName defines the table name of v1.SlaLifecycle type.
+var tableName = database.TableName(v1.NewSlaLifecycle())
+
 // GetCheckableFromSlaLifecycle returns the original checkable from which the specified sla lifecycle were transformed.
 // When the passed entity is not of type *SlaLifecycle, it is returned as is.
 func GetCheckableFromSlaLifecycle(e database.Entity) database.Entity {
@@ -86,3 +93,42 @@ func CreateSlaLifecyclesFromCheckables(
 
 	return slaLifecycles
 }
+
+// StreamIDsFromUpdatedSlaLifecycles updates the `delete_time` of the sla lifecycle table for each of the Checkables
+// consumed from the provided "entities" chan and upon successful execution of the query streams the original IDs
+// of the entities into the returned channel.
+//
+// It's unlikely, but when a given Checkable doesn't already have a `create_time` entry in the database, the update
+// query won't update anything. Either way the entities IDs are streamed into the returned chan.
+func StreamIDsFromUpdatedSlaLifecycles(
+	ctx context.Context, db *database.DB, g *errgroup.Group, logger *logging.Logger, entities <-chan database.Entity, bulkSize int,
+) <-chan any {
+	deleteEntityIDs := make(chan any, 1)
+
+	g.Go(func() error {
+		defer close(deleteEntityIDs)
+
+		var counter com.Counter
+		defer periodic.Start(ctx, logger.Interval(), func(_ periodic.Tick) {
+			if count := counter.Reset(); count > 0 {
+				logger.Infof("Updated %d sla lifecycles", count)
+			}
+		}).Stop()
+
+		sem := db.GetSemaphoreForTable(tableName)
+		stmt := fmt.Sprintf(`UPDATE %s SET delete_time = :delete_time WHERE "id" = :id AND "delete_time" = 0`, tableName)
+
+		if bulkSize <= 0 {
+			bulkSize = db.Options.MaxPlaceholdersPerStatement
+		}
+
+		// extractEntityId is used as a callback for the on success mechanism to extract the checkables id.
+		extractEntityId := func(e database.Entity) any { return e.(*v1.SlaLifecycle).SourceEntity.ID() }
+
+		return db.NamedBulkExec(
+			ctx, stmt, bulkSize, sem, CreateSlaLifecyclesFromCheckables(ctx, g, entities, true),
+			com.NeverSplit[database.Entity], OnSuccessApplyAndSendTo[database.Entity, any](deleteEntityIDs, extractEntityId))
+	})
+
+	return deleteEntityIDs
+}

pkg/icingadb/sync.go

@@ -184,27 +184,12 @@ func (s Sync) ApplyDelta(ctx context.Context, delta *Delta) error {
 		entity := delta.Subject.Entity()
 		switch entity.(type) {
 		case *v1.Host, *v1.Service:
-			deleteIds := make(chan any, len(delta.Delete))
-			g.Go(func() error {
-				defer close(deleteIds)
-
-				s.logger.Infof("Updating %d %s sla lifecycles", len(delta.Delete), delta.Subject.Name())
-
-				slTableName := database.TableName(v1.NewSlaLifecycle())
-				sem := s.db.GetSemaphoreForTable(slTableName)
-				stmt := fmt.Sprintf(`UPDATE %s SET delete_time = :delete_time WHERE "id" = :id AND "delete_time" = 0`, slTableName)
-
-				// extractEntityId is used as a callback for the on success mechanism to extract the checkables id.
-				extractEntityId := func(e database.Entity) any { return e.(*v1.SlaLifecycle).SourceEntity.ID() }
-
-				return s.db.NamedBulkExec(
-					ctx, stmt, s.db.Options.MaxPlaceholdersPerStatement, sem,
-					CreateSlaLifecyclesFromCheckables(ctx, g, delta.Delete.Entities(ctx), true),
-					com.NeverSplit[database.Entity], OnSuccessApplyAndSendTo[database.Entity, any](deleteIds, extractEntityId))
-			})
+			s.logger.Infof("Updating %d %s sla lifecycles", len(delta.Delete), delta.Subject.Name())
 
 			g.Go(func() error {
-				return s.db.DeleteStreamed(ctx, delta.Subject.Entity(), deleteIds, database.OnSuccessIncrement[any](stat))
+				return s.db.DeleteStreamed(
+					ctx, entity, StreamIDsFromUpdatedSlaLifecycles(ctx, s.db, g, s.logger, delta.Delete.Entities(ctx), 0),
+					database.OnSuccessIncrement[any](stat))
 			})
 		default:
 			g.Go(func() error {