engine/cleanupmgr: finalize migration to job queue for cleanup operat…

…ions (#4218)

* cleanup: implement context-aware sleep and cleanup manager functionality

* cleanup: format code for consistency in alert store initialization

* cleanup: remove unused ConfigSource field from SetupArgs struct

* cleanup: refactor alert cleanup logic and add shift cleanup functionality

* cleanup: refactor alert cleanup logic to use whileWork for better control flow

* cleanup: add comment

* cleanup: remove unused cleanup statements from DB and update logic

* test: refactor alert auto-close and cleanup tests for improved reliability

* fix: update ShiftArgs Kind to reflect cleanup-manager-shifts

* feat: add periodic jobs for schedule data cleanup and update queries

* feat: add workers for cleanup of shifts and schedule data

* fix: update periodic job interval for schedule data cleanup to 24 hours

* feat: add logging support to cleanup manager and engine initialization

* refactor: streamline schedule data cleanup by extracting user validation and shift trimming logic

* feat: add logging for schedule data updates in cleanup manager

* refactor: remove unnecessary checks for empty shifts in user and shift trimming functions

* refactor: enhance schedule data cleanup by improving user validation and logging

* refactor: improve formatting of schedule data update call in cleanup manager

* docs: add comment to clarify cleanupData function purpose in schedule data management

* feat: add CleanupAlertLogs function to manage alert log entries for deleted alerts

* refactor: remove unused cleanupAlertLogs statement and related logic from cleanup manager

* feat: implement timeout for CleanupAlertLogs worker to handle longer job durations

* refactor: rename cleanupDays to more descriptive staleThresholdDays in cleanup manager functions

* docs: enhance comment for CleanupMgrScheduleData to clarify last_cleanup_at usage

* engine/cleanupmgr: refactor schedule data cleanup logic and add job for looking up schedules needing cleanup

* engine/cleanupmgr: implement alert log cleanup job scheduling and refactor related queries

* engine/initriver: remove unused noopWorker type

* engine/cleanupmgr: rename LookForWorkArgs types for consistency

* feat(cleanupmanager): implement API key cleanup functionality

* feat(cleanupmanager): add periodic job for API key cleanup and remove unused update logic

* fix merge issue

* fix merge issue

* regen

* fix arg type

engine/cleanupmanager/apikeys.go

@@ -0,0 +1,50 @@
+package cleanupmanager
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/riverqueue/river"
+	"github.com/target/goalert/config"
+	"github.com/target/goalert/gadb"
+)
+
+type APIKeysArgs struct{}
+
+func (APIKeysArgs) Kind() string { return "cleanup-manager-api-keys" }
+
+// CleanupAPIKeys will revoke access to the API from unused tokens, including both user sessions and calendar subscriptions.
+func (db *DB) CleanupAPIKeys(ctx context.Context, j *river.Job[APIKeysArgs]) error {
+	err := db.whileWork(ctx, func(ctx context.Context, tx *sql.Tx) (done bool, err error) {
+		// After 30 days, the token is no longer valid, so delete it.
+		//
+		// This is defined by how the keyring system works for session signing, and is not influenced by the APIKeyExpireDays config.
+		count, err := gadb.New(tx).CleanupMgrDeleteOldSessions(ctx, 30)
+		if err != nil {
+			return false, fmt.Errorf("delete old user sessions: %w", err)
+		}
+		return count < 100, nil
+	})
+	if err != nil {
+		return err
+	}
+
+	cfg := config.FromContext(ctx)
+	if cfg.Maintenance.APIKeyExpireDays <= 0 {
+		return nil
+	}
+
+	err = db.whileWork(ctx, func(ctx context.Context, tx *sql.Tx) (done bool, err error) {
+		count, err := gadb.New(tx).CleanupMgrDisableOldCalSub(ctx, int32(cfg.Maintenance.APIKeyExpireDays))
+		if err != nil {
+			return false, fmt.Errorf("disable unused calsub keys: %w", err)
+		}
+		return count < 100, nil
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

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/queries.sql

@@ -189,3 +189,37 @@ _delete AS (
                     scope OFFSET @batch_size - 1
                 LIMIT 1;
 
+-- name: CleanupMgrDeleteOldSessions :execrows
+-- CleanupMgrDeleteOldSessions will delete old sessions from the auth_user_sessions table that are older than the given number of days before now.
+DELETE FROM auth_user_sessions
+WHERE id = ANY (
+        SELECT
+            id
+        FROM
+            auth_user_sessions
+        WHERE
+            last_access_at <(now() - '1 day'::interval * sqlc.arg(max_session_age_days)::int)
+        LIMIT 100
+        FOR UPDATE
+            SKIP LOCKED);
+
+-- name: CleanupMgrDisableOldCalSub :execrows
+-- CleanupMgrDeleteOldCalSub will disable old calendar subscriptions from the user_calendar_subscriptions table that are unused for at least the given number of days.
+UPDATE
+    user_calendar_subscriptions
+SET
+    disabled = TRUE
+WHERE
+    id = ANY (
+        SELECT
+            id
+        FROM
+            user_calendar_subscriptions
+        WHERE
+            greatest(last_access, last_update) <(now() - '1 day'::interval * sqlc.arg(inactivity_threshold_days)::int)
+        ORDER BY
+            id
+        LIMIT 100
+        FOR UPDATE
+            SKIP LOCKED);
+

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
 }