feat(cleanupmanager): add periodic job for API key cleanup and remove…

… unused update logic

engine/cleanupmanager/db.go

@@ -7,19 +7,13 @@ import (
 
 	"github.com/target/goalert/alert"
 	"github.com/target/goalert/engine/processinglock"
-	"github.com/target/goalert/util"
 )
 
 // DB handles updating escalation policies.
 type DB struct {
 	db   *sql.DB
 	lock *processinglock.Lock
 
-	cleanupAPIKeys *sql.Stmt
-	setTimeout     *sql.Stmt
-
-	cleanupSessions *sql.Stmt
-
 	alertStore *alert.Store
 
 	logger *slog.Logger
@@ -38,20 +32,11 @@ func NewDB(ctx context.Context, db *sql.DB, alertstore *alert.Store, log *slog.L
 		return nil, err
 	}
 
-	p := &util.Prepare{Ctx: ctx, DB: db}
-
 	return &DB{
 		db:     db,
 		lock:   lock,
 		logger: log,
 
-		// Abort any cleanup operation that takes longer than 3 seconds
-		// error will be logged.
-		setTimeout:     p.P(`SET LOCAL statement_timeout = 3000`),
-		cleanupAPIKeys: p.P(`update user_calendar_subscriptions set disabled = true where id = any(select id from user_calendar_subscriptions where greatest(last_access, last_update) < (now() - $1::interval) order by id limit 100 for update skip locked)`),
-
-		cleanupSessions: p.P(`DELETE FROM auth_user_sessions WHERE id = any(select id from auth_user_sessions where last_access_at < (now() - '30 days'::interval) LIMIT 100 for update skip locked)`),
-
 		alertStore: alertstore,
-	}, p.Err
+	}, nil
 }

engine/cleanupmanager/setup.go

@@ -18,6 +18,7 @@ const QueueName = "cleanup-manager"
 const (
 	PriorityAlertCleanup = 1
 	PrioritySchedHistory = 1
+	PriorityAPICleanup   = 1
 	PriorityTempSchedLFW = 2
 	PriorityAlertLogsLFW = 2
 	PriorityTempSched    = 3
@@ -57,6 +58,7 @@ func (db *DB) Setup(ctx context.Context, args processinglock.SetupArgs) error {
 	river.AddWorker(args.Workers, river.WorkFunc(db.LookForWorkScheduleData))
 	river.AddWorker(args.Workers, river.WorkFunc(db.CleanupAlertLogs))
 	river.AddWorker(args.Workers, river.WorkFunc(db.LookForWorkAlertLogs))
+	river.AddWorker(args.Workers, river.WorkFunc(db.CleanupAPIKeys))
 
 	err := args.River.Queues().Add(QueueName, river.QueueConfig{MaxWorkers: 5})
 	if err != nil {
@@ -117,5 +119,18 @@ func (db *DB) Setup(ctx context.Context, args processinglock.SetupArgs) error {
 		),
 	})
 
+	args.River.PeriodicJobs().AddMany([]*river.PeriodicJob{
+		river.NewPeriodicJob(
+			river.PeriodicInterval(24*time.Hour),
+			func() (river.JobArgs, *river.InsertOpts) {
+				return APIKeysArgs{}, &river.InsertOpts{
+					Queue:    QueueName,
+					Priority: PriorityAPICleanup,
+				}
+			},
+			&river.PeriodicJobOpts{RunOnStart: true},
+		),
+	})
+
 	return nil
 }