Retention Policies with Partitioning

api/dashboard_integration_test.go

@@ -3141,9 +3141,9 @@ func (s *ProjectIntegrationTestSuite) TestGetProjectStats() {
 	var stats *datastore.ProjectStatistics
 	parseResponse(s.T(), w.Result(), &stats)
 
-	require.Equal(s.T(), int64(2), stats.TotalEndpoints)
-	require.Equal(s.T(), int64(2), stats.TotalSources)
-	require.Equal(s.T(), int64(2), stats.TotalSubscriptions)
+	require.Equal(s.T(), true, stats.TotalEndpoints)     // int64(2)
+	require.Equal(s.T(), true, stats.TotalSources)       // int64(2)
+	require.Equal(s.T(), true, stats.TotalSubscriptions) // int64(2)
 }
 
 func (s *ProjectIntegrationTestSuite) TearDownTest() {

cmd/main.go

@@ -154,7 +154,7 @@ func main() {
 	c.AddCommand(ff.AddFeatureFlagsCommand())
 	c.AddCommand(utils.AddUtilsCommand(app))
 
-	if err := c.Execute(); err != nil {
+	if err = c.Execute(); err != nil {
 		slog.Fatal(err)
 	}
 }

cmd/server/server.go

@@ -156,9 +156,16 @@ func startConvoyServer(a *cli.App) error {
 	// register tasks
 	s.RegisterTask("58 23 * * *", convoy.ScheduleQueue, convoy.DeleteArchivedTasksProcessor)
 	s.RegisterTask("30 * * * *", convoy.ScheduleQueue, convoy.MonitorTwitterSources)
-	s.RegisterTask("0 0 * * *", convoy.ScheduleQueue, convoy.RetentionPolicies)
 	s.RegisterTask("0 * * * *", convoy.ScheduleQueue, convoy.TokenizeSearch)
 
+	// ensures that project data is backed up about 2 hours before they are deleted
+	if a.Licenser.RetentionPolicy() {
+		// runs at 10pm
+		s.RegisterTask("0 22 * * *", convoy.ScheduleQueue, convoy.BackupProjectData)
+		// runs at 1am
+		s.RegisterTask("0 1 * * *", convoy.ScheduleQueue, convoy.RetentionPolicies)
+	}
+
 	metrics.RegisterQueueMetrics(a.Queue, a.DB, nil)
 
 	km := keys.NewHCPVaultKeyManagerFromConfig(cfg.HCPVault, a.Licenser)

cmd/utils/partition.go

@@ -2,26 +2,133 @@ package utils
 
 import (
 	"fmt"
+	"github.com/frain-dev/convoy/config"
+	"github.com/frain-dev/convoy/database/postgres"
+	"github.com/frain-dev/convoy/internal/pkg/cli"
+	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/spf13/cobra"
 )
 
-func AddPartitionCommand() *cobra.Command {
+func AddPartitionCommand(a *cli.App) *cobra.Command {
+	var table string
+
 	cmd := &cobra.Command{
 		Use:   "partition",
 		Short: "runs partition commands",
 		Annotations: map[string]string{
 			"CheckMigration":  "true",
-			"ShouldBootstrap": "false",
+			"ShouldBootstrap": "true",
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
-			fmt.Println("Running partition command...")
+			cfg, err := config.Get()
+			if err != nil {
+				a.Logger.WithError(err).Fatal("Failed to load configuration")
+			}
+
+			featureFlag := fflag.NewFFlag(cfg.EnableFeatureFlag)
+			if !featureFlag.CanAccessFeature(fflag.RetentionPolicy) {
+				return fmt.Errorf("partitioning is only avaliable when the retention policy fflag is enabled")
+			}
+
+			if !a.Licenser.RetentionPolicy() {
+				return fmt.Errorf("partitioning is only avaliable with a license key")
+			}
+
+			eventsRepo := postgres.NewEventRepo(a.DB)
+			eventDeliveryRepo := postgres.NewEventDeliveryRepo(a.DB)
+			deliveryAttemptsRepo := postgres.NewDeliveryAttemptRepo(a.DB)
+
+			if table == "" {
+				return fmt.Errorf("table name is required")
+			}
+
+			switch table {
+			case "events":
+				err := eventsRepo.PartitionEventsTable(cmd.Context())
+				if err != nil {
+					return err
+				}
+			case "event-deliveries":
+				err := eventDeliveryRepo.PartitionEventDeliveriesTable(cmd.Context())
+				if err != nil {
+					return err
+				}
+			case "delivery-attempts":
+				err := deliveryAttemptsRepo.PartitionDeliveryAttemptsTable(cmd.Context())
+				if err != nil {
+					return err
+				}
+			default:
+				return fmt.Errorf("unknown table %s", table)
+			}
+
 			return nil
 		},
 	}
 
+	cmd.Flags().StringVarP(&table, "table", "t", "", "table name")
+
 	return cmd
 }
 
-func init() {
-	utilsCmd.AddCommand(AddPartitionCommand())
+func AddUnPartitionCommand(a *cli.App) *cobra.Command {
+	var table string
+
+	cmd := &cobra.Command{
+		Use:   "unpartition",
+		Short: "runs partition commands",
+		Annotations: map[string]string{
+			"CheckMigration":  "true",
+			"ShouldBootstrap": "true",
+		},
+		RunE: func(cmd *cobra.Command, args []string) error {
+			cfg, err := config.Get()
+			if err != nil {
+				a.Logger.WithError(err).Fatal("Failed to load configuration")
+			}
+
+			featureFlag := fflag.NewFFlag(cfg.EnableFeatureFlag)
+			if !featureFlag.CanAccessFeature(fflag.RetentionPolicy) {
+				return fmt.Errorf("partitioning is only avaliable when the retention policy fflag is enabled")
+			}
+
+			if !a.Licenser.RetentionPolicy() {
+				return fmt.Errorf("partitioning is only avaliable with a license key")
+			}
+
+			eventsRepo := postgres.NewEventRepo(a.DB)
+			eventDeliveryRepo := postgres.NewEventDeliveryRepo(a.DB)
+			deliveryAttemptsRepo := postgres.NewDeliveryAttemptRepo(a.DB)
+
+			if table == "" {
+				return fmt.Errorf("table name is required")
+			}
+
+			switch table {
+			case "events":
+				err := eventsRepo.UnPartitionEventsTable(cmd.Context())
+				if err != nil {
+					return err
+				}
+			case "event-deliveries":
+				err := eventDeliveryRepo.UnPartitionEventDeliveriesTable(cmd.Context())
+				if err != nil {
+					return err
+				}
+			case "delivery-attempts":
+				err := deliveryAttemptsRepo.UnPartitionDeliveryAttemptsTable(cmd.Context())
+				if err != nil {
+					return err
+				}
+			default:
+				return fmt.Errorf("unknown table %s", table)
+			}
+
+			return nil
+		},
+	}
+
+	cmd.Flags().StringVarP(&table, "table", "t", "", "table name")
+
+	return cmd
 }

cmd/utils/utils.go

@@ -15,10 +15,11 @@ var utilsCmd = &cobra.Command{
 }
 
 func AddUtilsCommand(app *cli.App) *cobra.Command {
+	utilsCmd.AddCommand(AddPartitionCommand(app))
+	utilsCmd.AddCommand(AddUnPartitionCommand(app))
 
 	utilsCmd.AddCommand(AddInitEncryptionCommand(app))
 	utilsCmd.AddCommand(AddRotateKeyCommand(app))
 	utilsCmd.AddCommand(AddRevertEncryptionCommand(app))
-
 	return utilsCmd
 }

cmd/worker/worker.go

@@ -7,8 +7,10 @@ import (
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/frain-dev/convoy/internal/pkg/keys"
+	"github.com/frain-dev/convoy/internal/pkg/retention"
 	"net/http"
 	"strings"
+	"time"
 
 	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/config"
@@ -325,6 +327,24 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		lo.Warn(fflag.ErrCircuitBreakerNotEnabled)
 	}
 
+	var ret retention.Retentioner
+	if featureFlag.CanAccessFeature(fflag.RetentionPolicy) && a.Licenser.RetentionPolicy() {
+		policy, _err := time.ParseDuration(cfg.RetentionPolicy.Policy)
+		if _err != nil {
+			lo.WithError(_err).Fatal("Failed to parse retention policy")
+			return _err
+		}
+
+		ret, err = retention.NewRetentionPolicy(a.DB, lo, policy)
+		if err != nil {
+			lo.WithError(err).Fatal("Failed to create retention policy")
+		}
+
+		ret.Start(ctx, time.Minute)
+	} else {
+		lo.Warn(fflag.ErrRetentionPolicyNotEnabled)
+	}
+
 	channels := make(map[string]task.EventChannel)
 	defaultCh, broadcastCh, dynamicCh := task.NewDefaultEventChannel(), task.NewBroadcastEventChannel(subscriptionsTable), task.NewDynamicEventChannel()
 	channels["default"] = defaultCh
@@ -390,14 +410,7 @@ func StartWorker(ctx context.Context, a *cli.App, cfg config.Configuration, inte
 		subRepo,
 		deviceRepo, a.Licenser), newTelemetry)
 
-	consumer.RegisterHandlers(convoy.RetentionPolicies, task.RetentionPolicies(
-		configRepo,
-		projectRepo,
-		eventRepo,
-		eventDeliveryRepo,
-		attemptRepo,
-		rd,
-	), nil)
+	consumer.RegisterHandlers(convoy.RetentionPolicies, task.RetentionPolicies(rd, ret), nil)
 
 	consumer.RegisterHandlers(convoy.MatchEventSubscriptionsProcessor, task.MatchSubscriptionsAndCreateEventDeliveries(
 		channels,