Add batch creation logic for the reminder service

cmd/reminder/app/start.go

@@ -49,11 +49,9 @@ func start(cmd *cobra.Command, _ []string) error {
 		return fmt.Errorf("unable to read config: %w", err)
 	}
 
-	// Configuration is normalized to ensure that all values are valid
-	// and if they can't be fixed, an error is returned
-	_, err = cfg.Normalize(cmd)
+	err = cfg.Validate()
 	if err != nil {
-		return fmt.Errorf("unable to normalize config: %w", err)
+		return fmt.Errorf("error validating config: %w", err)
 	}
 
 	ctx = logger.FromFlags(cfg.LoggingConfig).WithContext(ctx)

config/reminder-config.yaml.example

@@ -16,8 +16,6 @@
 recurrence:
   interval: "1h"
   batch_size: 100
-  max_per_project: 10
-  min_project_fetch_limit: 10
   min_elapsed: "1h"
 
 database:
@@ -28,8 +26,6 @@ database:
   dbname: minder
   sslmode: disable
 
-cursor_file: "/tmp/reminder-cursor"
-
 logging
   level: "debug"
 

database/migrations/000065_repo_reminder.down.sql

@@ -0,0 +1,15 @@
+-- Copyright 2024 Stacklok, Inc
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+ALTER TABLE repositories DROP COLUMN reminder_last_sent;

database/migrations/000065_repo_reminder.up.sql

@@ -0,0 +1,15 @@
+-- Copyright 2024 Stacklok, Inc
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+ALTER TABLE repositories ADD COLUMN reminder_last_sent TIMESTAMP;

database/migrations/000066_rule_evaluations_repo_index.down.sql

@@ -0,0 +1,15 @@
+-- Copyright 2024 Stacklok, Inc
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+DROP INDEX IF EXISTS rule_evaluations_repository_id_idx;

database/migrations/000066_rule_evaluations_repo_index.up.sql

@@ -0,0 +1,15 @@
+-- Copyright 2024 Stacklok, Inc
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+CREATE INDEX CONCURRENTLY rule_evaluations_repository_id_idx ON rule_evaluations(repository_id);

database/query/profile_status.sql

@@ -74,6 +74,16 @@ SELECT p.id, p.name, ps.profile_status, ps.last_updated FROM profile_status ps
 INNER JOIN profiles p ON p.id = ps.profile_id
 WHERE p.project_id = $1;
 
+-- ListOldestRuleEvaluationsByRepositoryId has casts in select statement as sqlc generates incorrect types.
+-- cast after MIN is required due to a known bug in sqlc: https://github.com/sqlc-dev/sqlc/issues/1965
+
+-- name: ListOldestRuleEvaluationsByRepositoryId :many
+SELECT re.repository_id::uuid AS repository_id, MIN(rde.last_updated)::timestamp AS oldest_last_updated
+FROM rule_evaluations re
+    INNER JOIN rule_details_eval rde ON re.id = rde.rule_eval_id
+WHERE re.repository_id = ANY (sqlc.arg('repository_ids')::uuid[])
+GROUP BY re.repository_id;
+
 -- DeleteRuleStatusesForProfileAndRuleType deletes a rule evaluation
 -- but locks the table before doing so.
 

database/query/repositories.sql

@@ -45,6 +45,25 @@ WHERE project_id = $1 AND webhook_id IS NOT NULL
     AND (lower(provider) = lower(sqlc.narg('provider')::text) OR sqlc.narg('provider')::text IS NULL)
 ORDER BY repo_name;
 
+-- name: ListRepositoriesAfterID :many
+SELECT *
+FROM repositories
+WHERE id > $1
+ORDER BY id
+LIMIT sqlc.arg('limit')::bigint;
+
+-- name: UpdateReminderLastSentById :exec
+UPDATE repositories
+SET reminder_last_sent = NOW()
+WHERE id = $1;
+
+-- name: RepositoryExistsAfterID :one
+SELECT EXISTS (
+  SELECT 1
+  FROM repositories
+  WHERE id > $1)
+AS exists;
+
 -- name: DeleteRepository :exec
 DELETE FROM repositories
 WHERE id = $1;

internal/config/reminder/config.go

@@ -16,11 +16,8 @@
 package reminder
 
 import (
-	"errors"
-	"fmt"
 	"strings"
 
-	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 
@@ -33,25 +30,10 @@ type Config struct {
 	RecurrenceConfig RecurrenceConfig      `mapstructure:"recurrence"`
 	EventConfig      EventConfig           `mapstructure:"events"`
 	LoggingConfig    LoggingConfig         `mapstructure:"logging"`
-	CursorFile       string                `mapstructure:"cursor_file" default:"/tmp/reminder_cursor"`
 }
 
-// Normalize normalizes the configuration
-// Returns a boolean indicating if the config was modified and an error if the config is invalid
-func (c *Config) Normalize(cmd *cobra.Command) (bool, error) {
-	err := c.validate()
-	if err != nil {
-		return c.patchConfig(cmd, err)
-	}
-
-	return false, nil
-}
-
-func (c *Config) validate() error {
-	if c == nil {
-		return errors.New("config cannot be nil")
-	}
-
+// Validate validates the configuration
+func (c Config) Validate() error {
 	err := c.RecurrenceConfig.Validate()
 	if err != nil {
 		return err
@@ -60,21 +42,6 @@ func (c *Config) validate() error {
 	return nil
 }
 
-func (c *Config) patchConfig(cmd *cobra.Command, err error) (bool, error) {
-	var batchSizeErr *InvalidBatchSizeError
-	if errors.As(err, &batchSizeErr) {
-		minAllowedBatchSize := batchSizeErr.MaxPerProject * batchSizeErr.MinProjectFetchLimit
-		cmd.Println("⚠ WARNING: " + batchSizeErr.Error())
-		cmd.Printf("Setting batch size to minimum allowed value: %d\n", minAllowedBatchSize)
-
-		// Update the config with the minimum allowed batch size
-		c.RecurrenceConfig.BatchSize = minAllowedBatchSize
-		return true, nil
-	}
-
-	return false, fmt.Errorf("invalid config: %w", err)
-}
-
 // SetViperDefaults sets the default values for the configuration to be picked up by viper
 func SetViperDefaults(v *viper.Viper) {
 	v.SetEnvPrefix("reminder")
@@ -84,13 +51,7 @@ func SetViperDefaults(v *viper.Viper) {
 
 // RegisterReminderFlags registers the flags for the minder cli
 func RegisterReminderFlags(v *viper.Viper, flags *pflag.FlagSet) error {
-	viperPath := "cursor_file"
-	if err := config.BindConfigFlag(v, flags, viperPath, "cursor-file",
-		v.GetString(viperPath), "DB Cursor file path for reminder", flags.String); err != nil {
-		return err
-	}
-
-	viperPath = "logging.level"
+	viperPath := "logging.level"
 	if err := config.BindConfigFlag(v, flags, viperPath, "logging-level",
 		v.GetString(viperPath), "Logging level for reminder", flags.String); err != nil {
 		return err