feat(retention): data retention

app/app.go

@@ -42,6 +42,7 @@ import (
 	"github.com/webhookx-io/webhookx/services"
 	"github.com/webhookx-io/webhookx/services/distributed"
 	"github.com/webhookx-io/webhookx/services/eventbus"
+	"github.com/webhookx-io/webhookx/services/retention"
 	"github.com/webhookx-io/webhookx/services/schedule"
 	"github.com/webhookx-io/webhookx/services/task"
 	tracingservice "github.com/webhookx-io/webhookx/services/tracing"
@@ -198,6 +199,16 @@ func (app *Application) initialize() error {
 		return err
 	}
 
+	if err := app.initRetention(&cfg.Retention, services); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (app *Application) initRetention(cfg *modules.RetentionConfig, services *services.Services) error {
+	retentionService := retention.NewRetentionService(*cfg, app.db, app.log, services.Scheduler)
+	app.registerService(retentionService)
 	return nil
 }
 
@@ -549,6 +560,10 @@ func (app *Application) Start() error {
 		app.log.Info("anonymous reports is disabled")
 	}
 
+	if app.cfg.Retention.Enabled && app.cfg.Role.IsDataPlane() {
+		app.log.Warnf("retention configuration is ignored on data-plane nodes")
+	}
+
 	services := []services.Service{
 		app.getService("eventbus"),
 		app.getService("metrics"),
@@ -558,6 +573,11 @@ func (app *Application) Start() error {
 		app.getService("status"),
 		app.getService("schedule"),
 	}
+
+	if !app.cfg.Role.IsDataPlane() {
+		services = append(services, app.getService("retention"))
+	}
+
 	return startServices(services...)
 }
 
@@ -582,7 +602,7 @@ func (app *Application) stop(ctx context.Context) error {
 	errs = append(errs, stopServices(ctx, 0, app.getService("eventbus")))
 	errs = append(errs, stopServices(ctx, 0, app.getService("worker")))
 	errs = append(errs, stopServices(ctx, time.Second*5, app.getService("metrics"), app.getService("tracing")))
-	errs = append(errs, stopServices(ctx, 0, app.getService("schedule")))
+	errs = append(errs, stopServices(ctx, 0, app.getService("schedule"), app.getService("retention")))
 	errs = append(errs, app.db.Close())
 
 	app.log.Info("exit")

config.yml

@@ -263,3 +263,18 @@ tracing:
 #      kubernetes:
 #        role:                                   # Vault role bound to the Kubernetes service account.
 #        token_path:                             # Path to JWT token file.
+
+
+#------------------------------------------------------------------------------
+# Data Retention
+#
+# Data Retention automatically remove expired objects from the database
+# periodically to reduce storage usage and improve query performance.
+#
+# This service only runs on control-plane or standalone nodes.
+# Data-plane nodes ignore this configuration.
+#------------------------------------------------------------------------------
+#retention:
+#  enabled: true                               # Whether to enable data retention. Defaults to false.
+#  events: 30                                  # Retention period (in days) for events. Set to 0 to disable event cleanup.
+#  attempts: 60                                # Retention period (in days) for delivery attempts. Set to 0 to disable attempt cleanup.

config/config.go

@@ -19,6 +19,10 @@ const (
 	RoleDPProxy    Role = "dp_proxy"
 )
 
+func (r Role) IsDataPlane() bool {
+	return r == RoleDPWorker || r == RoleDPProxy
+}
+
 var _ types.Config = &Config{}
 
 // Config Configuration
@@ -37,6 +41,7 @@ type Config struct {
 	Role             Role                    `yaml:"role" json:"role" envconfig:"ROLE" default:"standalone"`
 	AnonymousReports bool                    `yaml:"anonymous_reports" json:"anonymous_reports" envconfig:"ANONYMOUS_REPORTS" default:"true"`
 	Secret           modules.SecretConfig    `yaml:"secret" json:"secret" envconfig:"SECRET"`
+	Retention        modules.RetentionConfig `yaml:"retention" json:"retention" envconfig:"RETENTION"`
 }
 
 func (cfg *Config) PostProcess() error {
@@ -106,6 +111,9 @@ func (cfg Config) Validate() error {
 	if err := cfg.Secret.Validate(); err != nil {
 		return err
 	}
+	if err := cfg.Retention.Validate(); err != nil {
+		return err
+	}
 
 	return nil
 }

config/modules/retention.go

@@ -0,0 +1,13 @@
+package modules
+
+type RetentionConfig struct {
+	BaseConfig
+
+	Enabled  bool   `yaml:"enabled" json:"enabled" default:"false"`
+	Events   uint32 `yaml:"events" json:"events" default:"0"`
+	Attempts uint32 `yaml:"attempts" json:"attempts" default:"0"`
+}
+
+func (cfg RetentionConfig) Validate() error {
+	return nil
+}

db/dao/attempt_dao.go

@@ -93,6 +93,25 @@ func (dao *attemptDao) ListUnqueuedForUpdate(ctx context.Context, maxScheduledAt
 	return
 }
 
+func (dao *attemptDao) DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error) {
+	ctx, span := dao.trace(ctx, fmt.Sprintf("dao.%s.delete_ttl", dao.opts.Table))
+	defer span.End()
+
+	sql := fmt.Sprintf(
+		`DELETE FROM attempts WHERE id IN (
+			SELECT id FROM attempts WHERE created_at < now() - INTERVAL '%s' ORDER BY created_at ASC LIMIT $1
+		)`,
+		fmt.Sprintf("%d hours", int(ttl.Hours())),
+	)
+
+	dao.debugSQL(sql, []interface{}{limit})
+	res, err := dao.DB(ctx).ExecContext(ctx, sql, limit)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
 type AttemptQuery struct {
 	Query
 

db/dao/daos.go

@@ -36,6 +36,7 @@ type EventDAO interface {
 	BaseDAO[entities.Event]
 	BatchInsertIgnoreConflict(ctx context.Context, events []*entities.Event) ([]string, error)
 	ListExistingUniqueIDs(ctx context.Context, uniques []string) ([]string, error)
+	DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error)
 }
 
 type AttemptDAO interface {
@@ -44,6 +45,7 @@ type AttemptDAO interface {
 	UpdateErrorCode(ctx context.Context, id string, status entities.AttemptStatus, code entities.AttemptErrorCode) error
 	UpdateDelivery(ctx context.Context, result *AttemptResult) error
 	ListUnqueuedForUpdate(ctx context.Context, maxScheduledAt time.Time, limit int) (list []*entities.Attempt, err error)
+	DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error)
 }
 
 type SourceDAO interface {

db/dao/event_dao.go

@@ -3,6 +3,7 @@ package dao
 import (
 	"context"
 	"fmt"
+	"time"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/jmoiron/sqlx"
@@ -69,6 +70,24 @@ func (dao *eventDao) BatchInsertIgnoreConflict(ctx context.Context, events []*en
 	return inserteds, rows.Err()
 }
 
+func (dao *eventDao) DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error) {
+	ctx, span := dao.trace(ctx, fmt.Sprintf("dao.%s.delete_ttl", dao.opts.Table))
+	defer span.End()
+
+	sql := fmt.Sprintf(
+		`DELETE FROM events WHERE id IN (
+			SELECT id FROM events WHERE created_at < now() - INTERVAL '%s' ORDER BY created_at ASC LIMIT $1
+		)`,
+		fmt.Sprintf("%d hours", int(ttl.Hours())),
+	)
+
+	dao.debugSQL(sql, []interface{}{limit})
+	res, err := dao.DB(ctx).ExecContext(ctx, sql, limit)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
 
 type EventQuery struct {
 	Query

db/migrations/1785833137_retention.down.sql

@@ -0,0 +1,4 @@
+DROP INDEX IF EXISTS idx_events_created_at;
+DROP INDEX IF EXISTS idx_attempts_created_at;
+ALTER TABLE attempts ADD CONSTRAINT attempts_event_id_fkey FOREIGN KEY (event_id) REFERENCES events(id) ON DELETE CASCADE;
+ALTER TABLE attempts ADD CONSTRAINT attempts_endpoint_id_fkey FOREIGN KEY (endpoint_id) REFERENCES endpoints(id) ON DELETE CASCADE;

db/migrations/1785833137_retention.up.sql

@@ -0,0 +1,4 @@
+CREATE INDEX IF NOT EXISTS idx_events_created_at ON events (created_at);
+ALTER TABLE attempts DROP CONSTRAINT IF EXISTS attempts_endpoint_id_fkey;
+ALTER TABLE attempts DROP CONSTRAINT IF EXISTS  attempts_event_id_fkey;
+CREATE INDEX IF NOT EXISTS idx_attempts_created_at ON attempts (created_at);

services/retention/retention.go

@@ -0,0 +1,167 @@
+package retention
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/webhookx-io/webhookx/config/modules"
+	"github.com/webhookx-io/webhookx/db"
+	"github.com/webhookx-io/webhookx/services/schedule"
+	"github.com/webhookx-io/webhookx/utils"
+	"go.uber.org/zap"
+)
+
+const (
+	Interval  = time.Hour * 24
+	BatchSize = 1000
+
+	EventsKey   = "events"
+	AttemptsKey = "attempts"
+)
+
+type RetentionService struct {
+	cfg       modules.RetentionConfig
+	db        *db.DB
+	log       *zap.SugaredLogger
+	scheduler schedule.Scheduler
+	ttls      map[string]time.Duration
+}
+
+func NewRetentionService(
+	cfg modules.RetentionConfig,
+	db *db.DB,
+	log *zap.SugaredLogger,
+	scheduler schedule.Scheduler,
+) *RetentionService {
+
+	ttls := make(map[string]time.Duration)
+	ttls[EventsKey] = time.Hour * 24 * time.Duration(cfg.Events)
+	ttls[AttemptsKey] = time.Hour * 24 * time.Duration(cfg.Attempts)
+
+	return &RetentionService{
+		cfg:       cfg,
+		db:        db,
+		log:       log.Named("retention"),
+		scheduler: scheduler,
+		ttls:      ttls,
+	}
+}
+
+func (s *RetentionService) Name() string {
+	return "retention"
+}
+
+func (s *RetentionService) Start() error {
+	if !s.cfg.Enabled {
+		return nil
+	}
+
+	s.log.Infow("service started",
+		"purge_interval", utils.FormatDuration(Interval),
+		"batch_size", BatchSize,
+		zap.Any("ttl", map[string]string{
+			"events":   fmt.Sprintf("%dd", s.cfg.Events),
+			"attempts": fmt.Sprintf("%dd", s.cfg.Attempts),
+		}))
+
+	s.scheduler.Schedule(schedule.Task{
+		Name:      "retention",
+		Scheduled: schedule.NewIntervalSchedule(0, Interval),
+		Run: func(ctx context.Context) error {
+			return s.run(ctx)
+		},
+	})
+
+	return nil
+}
+
+func (s *RetentionService) Stop(ctx context.Context) error {
+	return nil
+}
+
+func (s *RetentionService) run(ctx context.Context) error {
+	if s.ttls[EventsKey] > 0 {
+		ttl := s.ttls[EventsKey]
+		count, err := s.purgeEvents(ctx, ttl)
+		if err != nil {
+			return err
+		}
+		if count > 0 {
+			s.log.Infof("deleted %d expired events", count)
+		}
+	}
+
+	if s.ttls[AttemptsKey] > 0 {
+		ttl := s.ttls[AttemptsKey]
+		count, err := s.purgeAttempts(ctx, ttl)
+		if err != nil {
+			return err
+		}
+		if count > 0 {
+			s.log.Infof("deleted %d expired attempts", count)
+		}
+	}
+
+	return nil
+}
+
+func (s *RetentionService) purgeEvents(ctx context.Context, ttl time.Duration) (int64, error) {
+	batchSize := BatchSize
+
+	s.log.Debugw("deleting expired events", "ttl", utils.FormatDuration(ttl), "batch_size", batchSize)
+
+	var total int64
+	for {
+		select {
+		case <-ctx.Done():
+			return total, ctx.Err()
+		default:
+		}
+
+		deleted, err := s.db.Events.DeleteTTL(ctx, ttl, batchSize)
+		if err != nil {
+			return total, fmt.Errorf("failed to delete expired events: %w", err)
+		}
+
+		total += deleted
+
+		if deleted < int64(batchSize) {
+			break
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	return total, nil
+}
+
+func (s *RetentionService) purgeAttempts(ctx context.Context, ttl time.Duration) (int64, error) {
+	batchSize := BatchSize
+
+	s.log.Debugw("deleting expired attempts", "ttl", utils.FormatDuration(ttl), "batch_size", batchSize)
+
+	var total int64
+	for {
+		select {
+		case <-ctx.Done():
+			return total, ctx.Err()
+		default:
+		}
+
+		deleted, err := s.db.Attempts.DeleteTTL(ctx, ttl, batchSize)
+		if err != nil {
+			return total, fmt.Errorf("failed to delete expired attempts: %w", err)
+		}
+
+		total += deleted
+
+		if deleted < int64(batchSize) {
+			break
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	return total, nil
+}

test/cmd/db_test.go

@@ -21,11 +21,12 @@ var statusOutputInit = `1 init (⏳ pending)
 10 ratelimit (⏳ pending)
 11 event_unique_id (⏳ pending)
 1762423418 source_config (⏳ pending)
+1785833137 retention (⏳ pending)
 Summary:
   Current version: 0
   Dirty: false
   Executed: 0
-  Pending: 12
+  Pending: 13
 `
 
 var statusOutputDone = `1 init (✅ executed)
@@ -40,10 +41,11 @@ var statusOutputDone = `1 init (✅ executed)
 10 ratelimit (✅ executed)
 11 event_unique_id (✅ executed)
 1762423418 source_config (✅ executed)
+1785833137 retention (✅ executed)
 Summary:
-  Current version: 1762423418
+  Current version: 1785833137
   Dirty: false
-  Executed: 12
+  Executed: 13
   Pending: 0
 `